bugfix for XAS only scans

debye_bec/devices/pilatus/pilatus.py

@@ -536,22 +536,27 @@ class Pilatus(PSIDeviceBase, ADBase):
 
     def on_pre_scan(self) -> DeviceStatus | None:
         """Called right before the scan starts on all devices automatically."""
-        status_hdf = CompareStatus(self.hdf.capture, ACQUIREMODE.ACQUIRING.value)
-        status_cam = CompareStatus(self.cam.acquire, ACQUIREMODE.ACQUIRING.value)
-        status_cam_server = CompareStatus(self.cam.armed, DETECTORSTATE.ARMED.value)
-        status = AndStatusWithList(
-            device=self, status_list=[status_hdf, status_cam, status_cam_server]
-        )
-        self.cam.acquire.put(1)
-        self.hdf.capture.put(1)
-        self.cancel_on_stop(status)
-        return status
+        scan_msg: ScanStatusMessage = self.scan_info.msg
+        if scan_msg.scan_name in self.xas_xrd_scan_names or scan_msg.scan_type == 'step':
+            status_hdf = CompareStatus(self.hdf.capture, ACQUIREMODE.ACQUIRING.value)
+            status_cam = CompareStatus(self.cam.acquire, ACQUIREMODE.ACQUIRING.value)
+            status_cam_server = CompareStatus(self.cam.armed, DETECTORSTATE.ARMED.value)
+            status = AndStatusWithList(
+                device=self, status_list=[status_hdf, status_cam, status_cam_server]
+            )
+            self.cam.acquire.put(1)
+            self.hdf.capture.put(1)
+            self.cancel_on_stop(status)
+            return status
+        else:
+            return None
 
     def on_trigger(self) -> DeviceStatus | None:
         """Called when the device is triggered."""
-        if self.scan_info.msg.scan_name in self.xas_xrd_scan_names:
+        scan_msg: ScanStatusMessage = self.scan_info.msg
+        if scan_msg.scan_name in self.xas_xrd_scan_names:
             return None
-        else:
+        elif scan_msg.scan_type == 'step':
             start_time = time.time()
             logger.warning(f"Triggering image with num_captured {self.hdf.num_captured.get()}")
             img_counter = self.hdf.num_captured.get()
@@ -560,42 +565,52 @@ class Pilatus(PSIDeviceBase, ADBase):
             self.trigger_shot.put(1)
             self.cancel_on_stop(status)
             return status
+        else:
+            return None
 
     def _complete_callback(self, status: DeviceStatus):
         """Callback for when the device completes a scan."""
-        if status.success:
-            status.device.file_event.put(
-                file_path=status.device._full_path,  # pylint: disable:protected-access
-                done=True,
-                successful=True,
-                hinted_h5_entries={"data": "/entry/data/data"},
-            )
+        scan_msg: ScanStatusMessage = self.scan_info.msg
+        if scan_msg.scan_name in self.xas_xrd_scan_names or scan_msg.scan_type == 'step':
+            if status.success:
+                status.device.file_event.put(
+                    file_path=status.device._full_path,  # pylint: disable:protected-access
+                    done=True,
+                    successful=True,
+                    hinted_h5_entries={"data": "/entry/data/data"},
+                )
+            else:
+                status.device.file_event.put(
+                    file_path=status.device._full_path,  # pylint: disable:protected-access
+                    done=True,
+                    successful=False,
+                    hinted_h5_entries={"data": "/entry/data/data"},
+                )
         else:
-            status.device.file_event.put(
-                file_path=status.device._full_path,  # pylint: disable:protected-access
-                done=True,
-                successful=False,
-                hinted_h5_entries={"data": "/entry/data/data"},
-            )
+            return None
 
     def on_complete(self) -> DeviceStatus | None:
         """Called to inquire if a device has completed a scans."""
-        status_hdf = CompareStatus(self.hdf.capture, ACQUIREMODE.DONE.value)
-        status_cam = CompareStatus(self.cam.acquire, ACQUIREMODE.DONE.value)
-        status_cam_server = CompareStatus(self.cam.armed, DETECTORSTATE.UNARMED.value)
-        if self.scan_info.msg.scan_name in self.xas_xrd_scan_names:
-            # For long scans, it can be that the mono will execute one cycle more,
-            # meaning a few more XRD triggers will be sent
-            status_img_written = CompareStatus(self.hdf.num_captured, self.n_images, operation='>=')
-        else:
+        scan_msg: ScanStatusMessage = self.scan_info.msg
+        if scan_msg.scan_name in self.xas_xrd_scan_names or scan_msg.scan_type == 'step':
+            status_hdf = CompareStatus(self.hdf.capture, ACQUIREMODE.DONE.value)
+            status_cam = CompareStatus(self.cam.acquire, ACQUIREMODE.DONE.value)
+            status_cam_server = CompareStatus(self.cam.armed, DETECTORSTATE.UNARMED.value)
+            if self.scan_info.msg.scan_name in self.xas_xrd_scan_names:
+                # For long scans, it can be that the mono will execute one cycle more,
+                # meaning a few more XRD triggers will be sent
+                status_img_written = CompareStatus(self.hdf.num_captured, self.n_images, operation='>=')
+            else:
+                status_img_written = CompareStatus(self.hdf.num_captured, self.n_images)
             status_img_written = CompareStatus(self.hdf.num_captured, self.n_images)
-        status_img_written = CompareStatus(self.hdf.num_captured, self.n_images)
-        status = AndStatusWithList(
-            device=self, status_list=[status_hdf, status_cam, status_img_written, status_cam_server]
-        )
-        status.add_callback(self._complete_callback)  # Callback that writing was successful
-        self.cancel_on_stop(status)
-        return status
+            status = AndStatusWithList(
+                device=self, status_list=[status_hdf, status_cam, status_img_written, status_cam_server]
+            )
+            status.add_callback(self._complete_callback)  # Callback that writing was successful
+            self.cancel_on_stop(status)
+            return status
+        else:
+            return None
 
     def on_kickoff(self) -> None:
         """Called to kickoff a device for a fly scan. Has to be called explicitly."""