diff --git a/superxas_bec/devices/timepix/timepix.py b/superxas_bec/devices/timepix/timepix.py index c8da464..6a8532b 100644 --- a/superxas_bec/devices/timepix/timepix.py +++ b/superxas_bec/devices/timepix/timepix.py @@ -15,6 +15,7 @@ import traceback from typing import TYPE_CHECKING, Any, Literal import numpy as np +from bec_lib.file_utils import get_full_path from bec_lib.logger import bec_logger from ophyd import ADBase from ophyd import Component as Cpt @@ -22,6 +23,7 @@ from ophyd_devices import ( AsyncSignal, CompareStatus, DeviceStatus, + FileEventSignal, PreviewSignal, StatusBase, TransitionStatus, @@ -41,7 +43,7 @@ from superxas_bec.devices.timepix.timepix_fly_client.timepix_fly_interface impor from superxas_bec.devices.timepix.utils import AndStatusWithList if TYPE_CHECKING: - from bec_lib.messages import DevicePreviewMessage + from bec_lib.messages import DevicePreviewMessage, ScanStatusMessage logger = bec_logger.logger @@ -124,6 +126,14 @@ class DATASOURCE(int, enum.Enum): IMAGE = 2 +class FILEWRITEMODE(int, enum.Enum): + """HDF5 Plugin FileWrite Mode""" + + SINGLE = 0 + CAPTURE = 1 + STREAM = 2 + + def load_pixel_map_from_json(file_path: str) -> PixelMap: """Load a pixel map from a JSON file. @@ -219,6 +229,10 @@ class Timepix(PSIDeviceBase, TimePixControl): doc="Preview signal of the TimePix detector.", ) + file_event = Cpt( + FileEventSignal, name="file_event", doc="File event signal for TimePix detector." + ) + def __init__( self, *, @@ -266,6 +280,7 @@ class Timepix(PSIDeviceBase, TimePixControl): 5 # Image poll rate for preview updates in Hz (max 5 Hz to limit throughput) ) self._enable_xes = enable_xes + self._full_path = "" self._unique_array_id = 0 self._pv_timeout = 5 self._readout_time = self.MIN_DETECTOR_READOUT_TIME @@ -511,6 +526,20 @@ class Timepix(PSIDeviceBase, TimePixControl): # Reset array counter on connect self.cam.array_counter.set(0).wait(timeout=self._pv_timeout) + # ------------------ + # Prepare file writing through AD HDF5 plugin + # ----------------- + self.hdf.enable.set(1).wait(timeout=self._pv_timeout) + self.hdf.file_write_mode.set(FILEWRITEMODE.STREAM.value).wait(timeout=self._pv_timeout) + self.hdf.auto_save.set(1).wait(timeout=self._pv_timeout) + self.hdf.file_template.set("%s%s").wait(timeout=self._pv_timeout) + self.hdf.lazy_open.set(1).wait(timeout=self._pv_timeout) + self.cam.array_callbacks.set(1).wait(timeout=self._pv_timeout) + + # ------------------ + # Prepare TimePixFly backend + # ----------------- + # Prepare backend for TimePixFly self.backend.on_connected() # Register the callback for processing data received by the backend @@ -522,12 +551,13 @@ class Timepix(PSIDeviceBase, TimePixControl): def on_stage(self) -> StatusBase | None: """Called while staging the device.""" - exp_time = self.scan_info.msg.scan_parameters.get("exp_time", 0) + scan_msg: ScanStatusMessage = self.scan_info.msg # type: ignore + exp_time = scan_msg.scan_parameters.get("exp_time", 0) if exp_time - self._readout_time <= 0: raise ValueError( f"Exposure time {exp_time} must be greater than readout time {self._readout_time}." ) - num_images = self.scan_info.msg.scan_parameters.get("frames_per_trigger", 1) + num_images = scan_msg.scan_parameters.get("frames_per_trigger", 1) logger.debug(f"Setting exposure time to {exp_time} and number of images to {num_images}") self.cam.acquire_time.set(exp_time - self._readout_time).wait(timeout=self._pv_timeout) @@ -535,6 +565,22 @@ class Timepix(PSIDeviceBase, TimePixControl): self.cam.num_images.set(num_images).wait(timeout=self._pv_timeout) self.cam.data_source.set(DATASOURCE.IMAGE).wait(timeout=self._pv_timeout) + # Setup file writing + self._full_path = get_full_path(scan_msg, name="timepix") + file_path = "/".join(self._full_path.split("/")[:-1]) + file_name = self._full_path.split("/")[-1] + self.cam.array_callbacks.set(1).wait(5) # Enable array callbacks + self.hdf.enable.set(1).wait(5) # Enable HDF5 plugin + self.hdf.file_path.set(file_path).wait(5) + self.hdf.file_name.set(file_name).wait(5) + self.hdf.num_capture.set(self.n_images).wait(5) + self.file_event.put( + file_path=self._full_path, + done=False, + successful=False, + hinted_h5_entries={"data": "/entry/data/data"}, + ) + # ------------------------- # XES specific staging if self.enable_xes: