bec/debye_bec
0
0
Fork 0
Code Issues 9 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

working example of combined xas_xrd scan

Browse Source
This commit is contained in:
gac-x01da
2025-09-11 09:19:51 +02:00
committed by appel_c
parent 199e52111f
commit bb25fb9c5f
1 changed files with 34 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

63
debye_bec/devices/pilatus/pilatus.py
View File

@@ -83,6 +83,13 @@ class TRIGGERMODE(int, enum.Enum):
MULT_TRIGGER = 3
ALIGNMENT = 4
class MONO_TRIGGER_SOURCE(int, enum.Enum):
""""Mono XRD trigger source"""
EPICS = 0
INPOS = 1
def description(self) -> str:
"""Return a description of the trigger mode."""
descriptions = {
@@ -210,6 +217,7 @@ class Pilatus(PSIDeviceBase, ADBase):
"xas_simple_scan_with_xrd",
"xas_advanced_scan_with_xrd",
]
self.n_images = None
# self._live_mode_thread = threading.Thread(
# target=self._live_mode_loop, daemon=True, name=f"{self.name}_live_mode_thread"
# )
@@ -226,7 +234,7 @@ class Pilatus(PSIDeviceBase, ADBase):
"""Poll the array data for preview updates."""
while not self._poll_thread_kill_event.wait(1 / self._poll_rate):
try:
logger.info(f"Running poll loop for {self.name}..")
# logger.info(f"Running poll loop for {self.name}..")
value = self.image1.array_data.get()
if value is None:
continue
@@ -235,7 +243,7 @@ class Pilatus(PSIDeviceBase, ADBase):
# Geometry correction for the image
data = np.reshape(value, (height, width))
last_image: DevicePreviewMessage = self.preview.get()
logger.info(f"Preview image for {self.name} has shape {data.shape}")
# logger.info(f"Preview image for {self.name} has shape {data.shape}")
if last_image is not None:
if np.array_equal(data, last_image.data):
# No update if image is the same, ~2.5ms on 2400x2400 image (6M)
@@ -443,12 +451,14 @@ class Pilatus(PSIDeviceBase, ADBase):
# logger.info(f'total trig low: {total_trig_lo}')
# logger.info(f'total trig high: {total_trig_hi}')
n_images = total_trig_lo + total_trig_hi
self.n_images = (total_trig_lo + total_trig_hi) * self.scan_parameter.n_of_trigger
exp_time = self.scan_parameter.exp_time
self.trigger_source.set(MONO_TRIGGER_SOURCE.INPOS).wait(5)
elif scan_msg.scan_type == 'step':
n_images = scan_msg.num_points * scan_msg.scan_parameters.get("frames_per_trigger", 1)
self.n_images = scan_msg.num_points * scan_msg.scan_parameters.get("frames_per_trigger", 1)
exp_time = scan_msg.scan_parameters.get("exp_time")
self.trigger_source.set(MONO_TRIGGER_SOURCE.EPICS).wait(5)
else:
return None
# Common settings
@@ -465,15 +475,15 @@ class Pilatus(PSIDeviceBase, ADBase):
self.hdf.enable.set(1).wait(5) # Enable HDF5 plugin
# Camera settings
self.cam.num_exposures.set(1).wait(5)
self.cam.num_images.set(n_images).wait(5)
self.cam.num_images.set(self.n_images).wait(5)
self.cam.acquire_time.set(detector_exp_time).wait(5) # let's try this
self.cam.acquire_period.set(exp_time).wait(5)
self.filter_number.set(0).wait(5)
# HDF5 settings
logger.debug(f"Setting HDF5 file path to {file_path} and file name to {file_name}")
logger.debug(f"Setting HDF5 file path to {file_path} and file name to {file_name}. full_path is {self._full_path}")
self.hdf.file_path.set(file_path).wait(5)
self.hdf.file_name.set(file_name).wait(5)
self.hdf.num_capture.set(n_images).wait(5)
self.hdf.num_capture.set(self.n_images).wait(5)
self.cam.array_counter.set(0).wait(5) # Reset array counter
self.file_event.put(
file_path=self._full_path,
@@ -487,26 +497,21 @@ class Pilatus(PSIDeviceBase, ADBase):
def on_pre_scan(self) -> DeviceStatus | None:
"""Called right before the scan starts on all devices automatically."""
if self.scan_info.msg.scan_name in self.xas_xrd_scan_names:
# TODO implement logic for 'xas' scans
return None
else:
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
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
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:
return None
# TODO implement logic for 'xas' scans
else:
start_time = time.time()
logger.warning(f"Triggering image with num_captured {self.hdf.num_captured.get()}")
@@ -536,16 +541,16 @@ class Pilatus(PSIDeviceBase, ADBase):
def on_complete(self) -> DeviceStatus | None:
"""Called to inquire if a device has completed a scans."""
if self.scan_info.msg.scan_name in self.xas_xrd_scan_names:
# TODO implement logic for 'xas' scans
return None
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)
num_images = self.scan_info.msg.num_points * self.scan_info.msg.scan_parameters.get(
"frames_per_trigger", 1
)
status_img_written = CompareStatus(self.hdf.num_captured, num_images)
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 = AndStatusWithList(
device=self, status_list=[status_hdf, status_cam, status_img_written, status_cam_server]
)