diff --git a/superxas_bec/device_configs/x10da_standard_config.yaml b/superxas_bec/device_configs/x10da_standard_config.yaml index 04808aa..8588c7d 100644 --- a/superxas_bec/device_configs/x10da_standard_config.yaml +++ b/superxas_bec/device_configs/x10da_standard_config.yaml @@ -67,6 +67,33 @@ nidaq: enabled: true softwareTrigger: false +### Timepix Detector +timepix: + readoutPriority: async + description: ASI Serval Timepix Detector + deviceClass: superxas_bec.devices.timepix.timepix.Timepix + deviceConfig: + prefix: "X10DA-ES-TPX1:" + backend_rest_url: "P6-0008.psi.ch:8452" + hostname: "x10da-bec-001.psi.ch" + enable_xes: false + onFailure: retry + enabled: true + readOnly: false + softwareTrigger: true + +### Falcon +falcon: + readoutPriority: monitored + description: Falcon Sitoro detector + deviceClass: superxas_bec.devices.falcon.FalconSuperXAS + deviceConfig: + prefix: 'X10DA-SITORO:' + onFailure: raise + enabled: True + softwareTrigger: False + + ## Remaining experimental hutch es_config: - - !include ./x10da_experimental_hutch.yaml \ No newline at end of file + - !include ./x10da_experimental_hutch.yaml diff --git a/superxas_bec/devices/timepix/timepix.py b/superxas_bec/devices/timepix/timepix.py index 5ebe319..d328c39 100644 --- a/superxas_bec/devices/timepix/timepix.py +++ b/superxas_bec/devices/timepix/timepix.py @@ -12,14 +12,20 @@ import os import threading import time import traceback +from collections import deque +from dataclasses import dataclass +from functools import partial 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 bec_server.scan_server.scans.scan_base import ScanInfo as ScanServerScanInfo from ophyd import ADBase from ophyd import Component as Cpt -from ophyd import Signal +from ophyd import Staged +from ophyd.status import StatusTimeoutError, WaitTimeoutError +from ophyd.utils import InvalidState from ophyd_devices import ( AsyncSignal, CompareStatus, @@ -44,6 +50,7 @@ from superxas_bec.devices.timepix.timepix_fly_client.timepix_fly_interface impor OtherConfigModel, PixelMap, ) +from superxas_bec.devices.utils.utils import fetch_scan_info if TYPE_CHECKING: from bec_lib.messages import ScanStatusMessage @@ -165,6 +172,27 @@ def load_pixel_map_from_json(file_path: str) -> PixelMap: return pixel_map +def window_fits(window: slice, length: int) -> bool: + """True if [start, stop) is a non-empty index window inside range(length).""" + return 0 <= window.start < window.stop <= length + + +def window_mean(data: np.ndarray, window: slice) -> float: + """ + Mean of data over the half-open index window [start, stop). + + Returns NaN if the window is empty or does not fit into data. numpy would instead average + over fewer points (stop past the end) or return NaN with a RuntimeWarning (empty window). + + Args: + data (np.ndarray): 1D data, e.g. one time spectrum. + window (slice): window with integer start and stop. + """ + if not window_fits(window, len(data)): + return float("nan") + return float(np.mean(data[window])) + + class ImagePlugin_Timepix(ImagePlugin_V35): """Custom Image Plugin for TimePix detector.""" @@ -192,6 +220,34 @@ DETECTOR_SHAPE = (512, 1024) # Shape of the TimePix detector TIMEPIX_FLY_BACKEND_DEFAULT_REDIS_TOPIC = "public/timepix_fly_backend/raw_data" +# scan_name of superxas_bec.scans.timepix_scan.TimepixScan +TIMEPIX_SCAN_NAME = "timepix_scan" + +# How long a repeated acquisition waits for the images of the lost one before it starts. +LOST_ACQUISITION_IMAGE_TIMEOUT = 2.0 + + +@dataclass +class XesAcquisition: + """ + One camera acquisition for which TimePixFly was armed. TimePixFly sends exactly one + StartFrame ... EndFrame sequence per armed acquisition, in order, so msg_buffer_callback + matches each EndFrame to the oldest pending acquisition. + + Attributes: + exposure (float): seconds the detector needs for the images of this trigger. + repeat_allowed (bool): the acquisition may be repeated if its raw data are lost. + acquire_time (float | None): time.monotonic() when Acquire was set, None before. + lost (bool): TimePixFly closed its frame before the exposure could end. + done (bool): the trigger status of this acquisition has completed. + """ + + exposure: float + repeat_allowed: bool + acquire_time: float | None = None + lost: bool = False + done: bool = False + class Timepix(PSIDeviceBase, TimePixControl): """ @@ -221,9 +277,18 @@ class Timepix(PSIDeviceBase, TimePixControl): static_spectra = Cpt(AsyncSignal, name="static_spectra", ndim=1, max_size=1000, acquisition_group="monitored", async_update={"type": "add", "max_shape": [None, DETECTOR_SHAPE[0]]}, doc="Spectra signal of the TimePix detector.") - # xes_data_accumulated_1 = Cpt(AsyncSignal, name="xes_accumulated_energy_1", ndim=1, max_size=1000, doc="1D time spectra for energy bin 2.") - # xes_data_accumulated_2 = Cpt(AsyncSignal, name="xes_accumulated_energy_2", ndim=1, max_size=1000, doc="1D time spectra for energy bin 2.") + xes_data_accumulated_1 = Cpt(AsyncSignal, name="xes_accumulated_energy_1", ndim=1, max_size=1000, doc="1D time spectra for energy bin 2.") + xes_data_accumulated_2 = Cpt(AsyncSignal, name="xes_accumulated_energy_2", ndim=1, max_size=1000, doc="1D time spectra for energy bin 2.") file_event = Cpt(FileEventSignal, name="file_event", doc="File event signal for TimePix detector.") + + ### Additional timepix scan signals + xes_level_1_off = Cpt(AsyncSignal, name="xes_level_1_off", ndim=0, async_update={"type": "add", "max_shape": [None]}, max_size=1000, doc="XES Level Off for energy bin 1.") + xes_level_1_on = Cpt(AsyncSignal, name="xes_level_1_on", ndim=0, async_update={"type": "add", "max_shape": [None]}, max_size=1000, doc="XES Level On for energy bin 1.") + xes_diff_1_on_off = Cpt(AsyncSignal, name="xes_diff_1_on_off", ndim=0, async_update={"type": "add", "max_shape": [None]}, max_size=1000, doc="XES Level Difference for energy bin 1.") + xes_level_2_off = Cpt(AsyncSignal, name="xes_level_2_off", ndim=0, async_update={"type": "add", "max_shape": [None]}, max_size=1000, doc="XES Level Off for energy bin 2.") + xes_level_2_on = Cpt(AsyncSignal, name="xes_level_2_on", ndim=0, async_update={"type": "add", "max_shape": [None]}, max_size=1000, doc="XES Level On for energy bin 2.") + xes_diff_2_on_off = Cpt(AsyncSignal, name="xes_diff_2_on_off", ndim=0, async_update={"type": "add", "max_shape": [None]}, max_size=1000, doc="XES Level Difference for energy bin 2.") + # fmt: on def __init__( @@ -259,6 +324,7 @@ class Timepix(PSIDeviceBase, TimePixControl): super().__init__( name=name, prefix=prefix, scan_info=scan_info, device_manager=device_manager, **kwargs ) + self.scan_parameters: ScanServerScanInfo | None = None self._redis_raw_data_topic = TIMEPIX_FLY_BACKEND_DEFAULT_REDIS_TOPIC self.backend = TimepixFlyBackend( @@ -269,8 +335,8 @@ class Timepix(PSIDeviceBase, TimePixControl): redis_connector=self.device_manager.connector, ) self._pixel_map = None - self._troistep = 1 - self._troin = 5000 + self._troistep = 1 # Will be changed if timepix scan is used, see on_stage call + self._troin = 5000 # Will be changed if timepix scan is used, see on_stage call self._poll_thread = threading.Thread( target=self._poll_array_data, daemon=True, name=f"{self.name}_poll_thread" @@ -288,13 +354,43 @@ class Timepix(PSIDeviceBase, TimePixControl): self.accumulated_data_e1 = None self.accumulated_data_e2 = None + # Settings that timepix_scan overrides in on_stage, saved for the restore in + # on_unstage / on_stop. None while no timepix_scan overrides are active. + self._settings_before_timepix_scan: dict[str, Any] | None = None + # Log the "on/off window does not fit" warning only once per scan. + self._on_off_window_warned = False + # Acquisitions armed in TimePixFly whose EndFrame has not arrived yet, oldest first. + self._xes_acquisitions: deque[XesAcquisition] = deque() + self._xes_acquisitions_lock = threading.Lock() + # Images of a lost acquisition still to arrive; they are not published. + self._images_to_skip = 0 + self._exposure_per_trigger = 0.0 + self._images_per_trigger = 1 def stage(self) -> list[object] | StatusBase: # type: ignore """Stage the device. - Super stage not safe to call..""" + Super stage not safe to call: ophyd's Device.stage would also stage the areaDetector + plugins and apply their stage_sigs. The device is still marked as staged, because the + BEC device server only calls unstage() (and with it on_unstage) on devices that report + Staged.yes. + """ self.stopped = False - status = self.on_stage() # pylint: disable=assignment-from-no-return + try: + status = self.on_stage() # pylint: disable=assignment-from-no-return + except Exception: + # Do not leave timepix_scan overrides behind if staging fails half-way. If the + # restore fails too, only log it so that the stage error is the one raised; the + # saved settings are then put back by the next stage, unstage or stop. + try: + self._restore_settings_after_timepix_scan() + except Exception: # pylint: disable=broad-except + content = traceback.format_exc() + logger.error( + f"Failed to restore settings of {self.name} after a failed stage: {content}" + ) + raise + self._staged = Staged.yes if isinstance(status, StatusBase): return status return [] @@ -355,6 +451,8 @@ class Timepix(PSIDeviceBase, TimePixControl): end_frame (dict): The EndFrame. Dictionary representation of detailed structure described in model .timepix_fly_client.timepix_fly_interface.TimepixEndFrame """ + if self._frame_of_lost_acquisition(end_frame): + return # on_trigger repeats the acquisition; publish nothing for this one n_energy_points = start_frame.get("NumEnergyPoints", None) troin = start_frame["TRoiN"] if troin != self._troin: @@ -407,27 +505,135 @@ class Timepix(PSIDeviceBase, TimePixControl): data_2 = np.sum(xes_data[:, 4:8], axis=1) self.xes_energy_1.put(data_1, async_update={"type": "add", "max_shape": [None, troin]}) self.xes_energy_2.put(data_2, async_update={"type": "add", "max_shape": [None, troin]}) - # if self.accumulated_data_e1 is None: - # self.accumulated_data_e1 = data_1 - # else: - # self.accumulated_data_e1 += data_1 - # if self.accumulated_data_e2 is None: - # self.accumulated_data_e2 = data_2 - # else: - # self.accumulated_data_e2 += data_2 - # self.xes_data_accumulated_1.put( - # self.accumulated_data_e1, - # async_update={"type": "replace", "max_shape": [None, troin]}, - # ) - # self.xes_data_accumulated_2.put( - # self.accumulated_data_e2, - # async_update={"type": "replace", "max_shape": [None, troin]}, - # ) + if self.accumulated_data_e1 is None: + self.accumulated_data_e1 = data_1 + else: + self.accumulated_data_e1 += data_1 + if self.accumulated_data_e2 is None: + self.accumulated_data_e2 = data_2 + else: + self.accumulated_data_e2 += data_2 + self.xes_data_accumulated_1.put( + self.accumulated_data_e1, + async_update={"type": "replace", "max_shape": [None, troin]}, + ) + self.xes_data_accumulated_2.put( + self.accumulated_data_e2, + async_update={"type": "replace", "max_shape": [None, troin]}, + ) + + ### Additional signals for timepix scan + # Frames can arrive without a staged timepix_scan (e.g. before the first stage after + # a restart), so scan_parameters may still be None here. + if self._is_timepix_scan(): + self._put_on_off_levels(data_1, data_2) + self.xes_spectra.put( xes_data.sum(axis=1), async_update={"type": "add", "max_shape": [None, troin]} ) logger.debug(f"Device data set for Timepix with {tds_period}, {tds_total_events}") + def _frame_of_lost_acquisition(self, end_frame: dict) -> bool: + """ + Match the EndFrame to its acquisition and check whether that acquisition was lost. + + A frame without any TDC period holds no XES data. If it arrives before the exposure of + its acquisition can have ended, TimePixFly's raw-data connection closed before the + detector measured, and the raw data of that acquisition never reached TimePixFly + (scan S00437, point 1: the frame closed 1.2 s before the detector delivered the image). + Such an acquisition is marked lost and on_trigger repeats it once, if allowed. A frame + without TDC periods after the exposure means the detector recorded no TDC pulse; it is + published as it is. + + Args: + end_frame (dict): EndFrame received from TimePixFly. + + Returns: + bool: True if the frame belongs to a lost acquisition that will be repeated. + """ + with self._xes_acquisitions_lock: + if not self._xes_acquisitions: + return False # no acquisition armed by this device, e.g. after a stop + acquisition = self._xes_acquisitions.popleft() + if end_frame.get("periods", None) != 0: + return False + if acquisition.acquire_time is None: + timing = "before Acquire was set" + else: + elapsed = time.monotonic() - acquisition.acquire_time + timing = f"{elapsed:.3f} s after Acquire was set" + if elapsed >= acquisition.exposure: + logger.error( + f"{self.name}: TimePixFly frame without any TDC period, {timing} " + f"({acquisition.exposure:.3f} s exposure). The detector recorded no TDC " + "pulse: check the TDC input signal. This point has no XES data." + ) + return False + if not acquisition.repeat_allowed or acquisition.done: + logger.error( + f"{self.name}: TimePixFly closed the frame {timing}, before the " + f"{acquisition.exposure:.3f} s exposure could end, without any TDC period. " + "The raw data of this point did not reach TimePixFly and the acquisition " + "cannot be repeated. This point has no XES data." + ) + return False + acquisition.lost = True + self._images_to_skip += self._images_per_trigger + logger.warning( + f"{self.name}: TimePixFly closed the frame {timing}, before the " + f"{acquisition.exposure:.3f} s exposure could end, without any TDC period. The raw " + "data of this acquisition did not reach TimePixFly; repeating the acquisition." + ) + return True + + def _put_on_off_levels(self, data_1: np.ndarray, data_2: np.ndarray) -> None: + """ + Publish the laser on/off levels of a timepix_scan for both energy groups. + + The off/on windows are the half-open time-bin ranges [idx_*_start, idx_*_stop) given as + timepix_scan arguments. A level is the mean of xes_energy_1/2 over its window, the + difference is on minus off. If a window is empty or does not fit into the received + time bins (TimePixFly sent fewer bins than time_roi_n), its levels and the differences + are published as NaN, so every frame still adds one value per signal and the signals stay + aligned with the scan points; a warning is logged once per scan. + + Args: + data_1 (np.ndarray): xes_energy_1 of this frame, one value per time bin. + data_2 (np.ndarray): xes_energy_2 of this frame, one value per time bin. + """ + params = self.scan_parameters.additional_scan_parameters + off = slice(params["idx_off_start"], params["idx_off_stop"]) + on = slice(params["idx_on_start"], params["idx_on_stop"]) + + bad_windows = [ + f"{name} [{window.start}, {window.stop})" + for name, window in (("off", off), ("on", on)) + if not window_fits(window, len(data_1)) + ] + if bad_windows and not self._on_off_window_warned: + self._on_off_window_warned = True + logger.warning( + f"{self.name}: laser {' and '.join(bad_windows)} window does not fit into the " + f"{len(data_1)} time bins received; publishing NaN for the affected levels." + ) + + level_1_off = window_mean(data_1, off) + level_1_on = window_mean(data_1, on) + level_2_off = window_mean(data_2, off) + level_2_on = window_mean(data_2, on) + self.xes_level_1_off.put(level_1_off) + self.xes_level_1_on.put(level_1_on) + self.xes_level_2_off.put(level_2_off) + self.xes_level_2_on.put(level_2_on) + self.xes_diff_1_on_off.put(level_1_on - level_1_off) + self.xes_diff_2_on_off.put(level_2_on - level_2_off) + + def _is_timepix_scan(self) -> bool: + """True if the device is staged for a timepix_scan (False before the first stage).""" + return ( + self.scan_parameters is not None and self.scan_parameters.scan_name == TIMEPIX_SCAN_NAME + ) + ### User ACCESS methods def get_pixel_map(self) -> dict: @@ -603,6 +809,11 @@ class Timepix(PSIDeviceBase, TimePixControl): """Callback for image unique ID updates to trigger preview update.""" if value == old_value: return # No new image, or counter reset + with self._xes_acquisitions_lock: + if self._images_to_skip > 0: + # Image of an acquisition whose XES data were lost; its repetition replaces it. + self._images_to_skip -= 1 + return try: # Get new image data width = self.image.array_size.width.get() @@ -621,9 +832,17 @@ class Timepix(PSIDeviceBase, TimePixControl): def on_stage(self) -> StatusBase | None: """Called while staging the device.""" - ### + self.scan_parameters = fetch_scan_info(self.scan_info) + + if self._is_timepix_scan(): + self._apply_timepix_scan_settings(self.scan_parameters.additional_scan_parameters) + else: + # Overrides of an earlier timepix_scan whose restore failed (no-op otherwise) + self._restore_settings_after_timepix_scan() + self.accumulated_data_e1 = None self.accumulated_data_e2 = None + self._on_off_window_warned = False 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: @@ -632,6 +851,9 @@ class Timepix(PSIDeviceBase, TimePixControl): ) burst_images = scan_msg.scan_parameters.get("frames_per_trigger", 1) self._n_images = scan_msg.num_points * burst_images + self._images_per_trigger = burst_images + self._exposure_per_trigger = burst_images * exp_time - self._readout_time + self._clear_xes_acquisitions() # Camera has to be set to burst_images, each step will get an individual trigger self.cam.acquire_time.set(exp_time - self._readout_time).wait(timeout=self._pv_timeout) self.cam.acquire_period.set(exp_time).wait(timeout=self._pv_timeout) @@ -680,12 +902,53 @@ class Timepix(PSIDeviceBase, TimePixControl): timeout=self._pv_timeout ) + def _apply_timepix_scan_settings(self, params: dict) -> None: + """ + Apply the timepix_scan overrides: time binning from the scan arguments, HDF5 image + writing off and XES on. The previous values are saved and put back by + _restore_settings_after_timepix_scan (on_unstage, or on_stop for an aborted scan). + + Args: + params (dict): additional_scan_parameters of the timepix_scan. + """ + if self._settings_before_timepix_scan is None: + # If an earlier restore failed, keep the values saved before that scan. + self._settings_before_timepix_scan = { + "troistep": self._troistep, + "troin": self._troin, + "hdf_enable": self.hdf.enable.get(), + "enable_xes": self.enable_xes, + } + self._troistep = params["time_roi_step"] + self._troin = params["time_roi_n"] + # Wait for the readback: the rest of on_stage checks hdf.enable to decide on capture. + self.set_enable_image_writing(False) + self.set_enable_xes(True) + + def _restore_settings_after_timepix_scan(self) -> None: + """ + Put back the settings saved by _apply_timepix_scan_settings. Does nothing if no + timepix_scan overrides are active, so on_stop, on_unstage and on_stage can all call it. + """ + settings = self._settings_before_timepix_scan + if settings is None: + return + self._troistep = settings["troistep"] + self._troin = settings["troin"] + self.hdf.enable.set(settings["hdf_enable"]).wait(timeout=self._pv_timeout) + self.set_enable_xes(settings["enable_xes"]) + # Clear only after success, so a failed restore in on_stop is retried in on_unstage. + self._settings_before_timepix_scan = None + def on_unstage(self) -> None: """Called while unstaging the device.""" - if self.cam.acquire_busy.get() == ACQUIRESTATUS.ACQUIRING: - self.cam.acquire.put(0) - status_camera = CompareStatus(self.cam.acquire_busy, ACQUIRESTATUS.DONE) - status_camera.wait(timeout=self._pv_timeout) + try: + if self.cam.acquire_busy.get() == ACQUIRESTATUS.ACQUIRING: + self.cam.acquire.put(0) + status_camera = CompareStatus(self.cam.acquire_busy, ACQUIRESTATUS.DONE) + status_camera.wait(timeout=self._pv_timeout) + finally: + self._restore_settings_after_timepix_scan() def on_pre_scan(self) -> StatusBase: """Called right before the scan starts on all devices automatically.""" @@ -705,51 +968,155 @@ class Timepix(PSIDeviceBase, TimePixControl): return status def on_trigger(self) -> DeviceStatus | StatusBase | None: - """Called when the device is triggered.""" + """ + Called when the device is triggered. - status_camera = TransitionStatus( + With XES enabled, TimePixFly is armed before the camera starts. If TimePixFly closes the + frame of the acquisition before its exposure could end (see _frame_of_lost_acquisition), + the acquisition is repeated once and the returned status resolves after the repetition. + This needs the HDF5 plugin to be off (as in timepix_scan): with image writing on, a + repetition would add an image to the file, so the point is kept without XES data. + """ + if self.enable_xes is not True: + status_camera = self._camera_acquisition_status() + self.cancel_on_stop(status_camera) + self.cam.acquire.put(1) + return status_camera + + point_status = DeviceStatus(self) + self.cancel_on_stop(point_status) + repeat_allowed = self.hdf.enable.get() != "Enable" + self._start_xes_acquisition(point_status, repeat_allowed=repeat_allowed) + return point_status + + def _camera_acquisition_status(self) -> TransitionStatus: + """Status that resolves once the camera has acquired (AcquireBusy DONE -> ACQUIRING -> DONE).""" + return TransitionStatus( self.cam.acquire_busy, [ACQUIRESTATUS.DONE, ACQUIRESTATUS.ACQUIRING, ACQUIRESTATUS.DONE] ) - status_backend = None + + def _start_xes_acquisition(self, point_status: DeviceStatus, repeat_allowed: bool) -> None: + """ + Arm TimePixFly and start one camera acquisition for the trigger status point_status. + _on_xes_acquisition_done resolves point_status, or repeats a lost acquisition. + + Args: + point_status (DeviceStatus): trigger status returned by on_trigger. + repeat_allowed (bool): the acquisition may be repeated if its raw data are lost. + + Raises: + TimeoutError: if TimePixFly does not reach the 'config' or 'await_connection' state. + """ + status_camera = self._camera_acquisition_status() # First we make sure that the backend reach 'config' state. This needs to happend before each trigger. - if self.enable_xes is True: - status_backend_config = DeviceStatus(self) - self.cancel_on_stop(status_backend_config) - self.backend.timepix_fly_client.add_status_callback( - status=status_backend_config, - success=[TimePixFlyStatus.CONFIG], - error=[TimePixFlyStatus.EXCEPT, TimePixFlyStatus.SHUTDOWN], - ) - try: - status_backend_config.wait(timeout=5) - except TimeoutError: - # pylint:disable=raise-missing-from - raise TimeoutError( - f"TimePixFly backend of device {self.name} failed to reach 'config' state in trigger" - ) - - # Prepare backend to be ready to receive trigger - status = self.backend.on_trigger() - try: - status.wait(timeout=5) # Wait until backend trigger is done + status_backend_config = DeviceStatus(self) + self.cancel_on_stop(status_backend_config) + self.backend.timepix_fly_client.add_status_callback( + status=status_backend_config, + success=[TimePixFlyStatus.CONFIG], + error=[TimePixFlyStatus.EXCEPT, TimePixFlyStatus.SHUTDOWN], + ) + try: + status_backend_config.wait(timeout=5) + except TimeoutError: # pylint:disable=raise-missing-from - except Exception: - logger.error( - f"TimePixFly backend of device {self.name} failed to prepare for trigger" - ) - raise TimeoutError( - f"TimePixFly backend of device {self.name} failed to reach 'trigger' state in trigger" - ) - # Status that resolves once the trigger is done - status_backend = self.backend.on_trigger_finished() + raise TimeoutError( + f"TimePixFly backend of device {self.name} failed to reach 'config' state in trigger" + ) - if status_backend is not None: - return_status = status_camera & status_backend - else: - return_status = status_camera - self.cancel_on_stop(return_status) + # Registered before arming: the EndFrame of a lost acquisition can arrive at once. + acquisition = XesAcquisition( + exposure=self._exposure_per_trigger, repeat_allowed=repeat_allowed + ) + with self._xes_acquisitions_lock: + self._xes_acquisitions.append(acquisition) + # Prepare backend to be ready to receive trigger + status = self.backend.on_trigger() + try: + status.wait(timeout=5) # Wait until backend trigger is done + # pylint:disable=raise-missing-from + except Exception: + with self._xes_acquisitions_lock: + if acquisition in self._xes_acquisitions: + self._xes_acquisitions.remove(acquisition) + logger.error(f"TimePixFly backend of device {self.name} failed to prepare for trigger") + raise TimeoutError( + f"TimePixFly backend of device {self.name} failed to reach 'trigger' state in trigger" + ) + # Status that resolves once the trigger is done + status_backend = self.backend.on_trigger_finished() + acquisition_status = status_camera & status_backend + self.cancel_on_stop(acquisition_status) + acquisition_status.add_callback( + partial(self._on_xes_acquisition_done, point_status, acquisition) + ) + acquisition.acquire_time = time.monotonic() self.cam.acquire.put(1) - return return_status + + def _on_xes_acquisition_done( + self, point_status: DeviceStatus, acquisition: XesAcquisition, status: StatusBase + ) -> None: + """Resolve the trigger status, or repeat the acquisition if TimePixFly lost its data.""" + with self._xes_acquisitions_lock: + acquisition.done = True + lost = acquisition.lost + if point_status.done: + return # cancelled, e.g. by a stop + if not status.success: + self._resolve_point_status(point_status, status.exception()) + return + if lost: + threading.Thread( + target=self._repeat_xes_acquisition, + args=(point_status,), + daemon=True, + name=f"{self.name}_repeat_acquisition", + ).start() + return + self._resolve_point_status(point_status) + + def _repeat_xes_acquisition(self, point_status: DeviceStatus) -> None: + """Repeat a lost acquisition once, after the images of the lost one have arrived.""" + try: + deadline = time.monotonic() + LOST_ACQUISITION_IMAGE_TIMEOUT + while True: + with self._xes_acquisitions_lock: + if self._images_to_skip == 0: + break + if time.monotonic() > deadline: + logger.warning( + f"{self.name}: {self._images_to_skip} image(s) of the lost " + "acquisition did not arrive; repeating without waiting for them." + ) + self._images_to_skip = 0 + break + time.sleep(0.05) + if point_status.done or self.stopped: + return + self._start_xes_acquisition(point_status, repeat_allowed=False) + except Exception as exc: # pylint: disable=broad-except + content = traceback.format_exc() + logger.error(f"{self.name}: repeating the lost acquisition failed: {content}") + self._resolve_point_status(point_status, exc) + + @staticmethod + def _resolve_point_status(point_status: DeviceStatus, exc: Exception | None = None) -> None: + """Finish point_status (failed if exc is given), unless it is already done.""" + if isinstance(exc, (StatusTimeoutError, WaitTimeoutError)): + exc = TimeoutError(str(exc)) # ophyd reserves these for its own timeouts + try: + if exc is None: + point_status.set_finished() + else: + point_status.set_exception(exc) + except InvalidState: + pass # resolved meanwhile, e.g. cancelled by a stop + + def _clear_xes_acquisitions(self) -> None: + """Forget pending acquisitions and images to skip (new scan, or stop).""" + with self._xes_acquisitions_lock: + self._xes_acquisitions.clear() + self._images_to_skip = 0 def on_complete(self) -> DeviceStatus | StatusBase | None: """Called to inquire if a device has completed a scans.""" @@ -813,6 +1180,16 @@ class Timepix(PSIDeviceBase, TimePixControl): # Backend if self.enable_xes is True: self.backend.on_stop() + self._clear_xes_acquisitions() + # BEC does not unstage the devices of an aborted scan, so undo the timepix_scan + # overrides here. Runs after the backend stop, which depends on enable_xes. + try: + self._restore_settings_after_timepix_scan() + except Exception: # pylint: disable=broad-except + content = traceback.format_exc() + logger.error( + f"Failed to restore settings of {self.name} after timepix_scan, will retry on unstage: {content}" + ) def on_destroy(self): """Cleanup method to stop the device and clean up resources.""" diff --git a/superxas_bec/scans/__init__.py b/superxas_bec/scans/__init__.py index 5f4a83b..078ccff 100644 --- a/superxas_bec/scans/__init__.py +++ b/superxas_bec/scans/__init__.py @@ -3,3 +3,4 @@ from .mono_bragg_scans import ( # XASAdvancedScanWithXRD,; XASSimpleScanWithXRD XASSimpleScan, ) from .nidaq_cont_scan import NIDAQContinuousScan +from .timepix_scan import TimepixScan diff --git a/superxas_bec/scans/timepix_scan.py b/superxas_bec/scans/timepix_scan.py new file mode 100644 index 0000000..f2bcced --- /dev/null +++ b/superxas_bec/scans/timepix_scan.py @@ -0,0 +1,95 @@ +""" +Timepix Scan (extension of line scan). + +Scan procedure: + - prepare_scan + - open_scan + - stage + - pre_scan + - scan_core + - at_each_point (optionally called by scan_core) + - post_scan + - unstage + - close_scan + - on_exception (called if any exception is raised during the scan) +""" + +from __future__ import annotations + +from typing import Annotated + +from bec_lib.scan_args import DefaultArgType, ScanArgument +from bec_server.scan_server.scans.line_scan import LineScan + + +class TimepixScan(LineScan): + """ + Timepix Scan + Extends Line Scan with additional parameters for the timepix detector. + """ + + scan_name = "timepix_scan" + + gui_config = { + "Movement Parameters": ["steps", "relative"], + "Acquisition Parameters": ["exp_time", "burst_at_each_point"], + "Binning Parameters": ["time_roi_step", "time_roi_n"], + "Laser On/Off Parameters": ["idx_off_start", "idx_off_stop", "idx_on_start", "idx_on_stop"], + } + + def __init__( + self, + *args, + steps: Annotated[int, ScanArgument(display_name="Number of Steps", gt=0)], + relative: DefaultArgType.Relative, + exp_time: DefaultArgType.ExposureTime = 0, + burst_at_each_point: DefaultArgType.BurstAtEachPoint = 1, + # fmt: off + time_roi_step: Annotated[int, ScanArgument(display_name="Time ROI step", description="Bin size", gt=0)], + time_roi_n: Annotated[int, ScanArgument(display_name="Time ROI n", description="Number of time points", gt=0)], + idx_off_start: Annotated[int, ScanArgument(display_name="Index Off Start", description="Start Index for Off Level (first time bin)", ge=0)], + idx_off_stop: Annotated[int, ScanArgument(display_name="Index Off Stop", description="Stop Index for Off Level (exclusive)", ge=0)], + idx_on_start: Annotated[int, ScanArgument(display_name="Index On Start", description="Start Index for On Level (first time bin)", ge=0)], + idx_on_stop: Annotated[int, ScanArgument(display_name="Index On Stop", description="Stop Index for On Level (exclusive)", ge=0)], + # fmt: on + **kwargs, + ): + # The on/off windows are half-open index ranges [start, stop) over the time_roi_n time + # bins; the Timepix device averages each window into xes_level_*. An empty window + # (start == stop) would only produce NaN, so it is rejected here as well. + if idx_off_start >= idx_off_stop: + raise ValueError( + f"idx_off_start ({idx_off_start}) must be smaller than idx_off_stop ({idx_off_stop})!" + ) + if idx_on_start >= idx_on_stop: + raise ValueError( + f"idx_on_start ({idx_on_start}) must be smaller than idx_on_stop ({idx_on_stop})!" + ) + for name, idx in ( + ("idx_off_start", idx_off_start), + ("idx_off_stop", idx_off_stop), + ("idx_on_start", idx_on_start), + ("idx_on_stop", idx_on_stop), + ): + if idx > time_roi_n: + raise ValueError( + f"{name} with value {idx} exceeds length of data array of time_roi_n of {time_roi_n}" + ) + + super().__init__( + *args, + steps=steps, + relative=relative, + exp_time=exp_time, + burst_at_each_point=burst_at_each_point, + **kwargs, + ) + + self.update_scan_info( + time_roi_step=time_roi_step, + time_roi_n=time_roi_n, + idx_off_start=idx_off_start, + idx_off_stop=idx_off_stop, + idx_on_start=idx_on_start, + idx_on_stop=idx_on_stop, + ) diff --git a/tests/tests_devices/test_timepix.py b/tests/tests_devices/test_timepix.py index 0010025..c7afa1f 100644 --- a/tests/tests_devices/test_timepix.py +++ b/tests/tests_devices/test_timepix.py @@ -3,14 +3,19 @@ from __future__ import annotations import threading -from types import SimpleNamespace +import time +from contextlib import ExitStack from unittest import mock import numpy as np import ophyd import pytest +from bec_lib.devicemanager import ScanInfo +from bec_lib.messages import ScanStatusMessage from bec_server.device_server.tests.utils import DMMock -from ophyd import DeviceStatus, StatusBase +from bec_server.scan_server.scans.scan_base import ScanInfo as ScanServerScanInfo +from ophyd import DeviceStatus, Staged, StatusBase +from ophyd_devices.interfaces.base_classes.psi_device_base import DeviceStoppedError from ophyd_devices.tests.utils import MockPV, patch_dual_pvs from superxas_bec.devices.timepix.timepix import ( @@ -23,12 +28,14 @@ from superxas_bec.devices.timepix.timepix import ( TDCEdge, TDCOuput, Timepix, + window_mean, ) from superxas_bec.devices.timepix.timepix_fly_client.timepix_fly_client import TimePixFlyStatus from superxas_bec.devices.timepix.timepix_fly_client.timepix_fly_interface import ( NetAddresses, PixelMap, ) +from superxas_bec.devices.utils.utils import fetch_scan_info # pylint: disable=protected-access # pylint: disable=redefined-outer-name @@ -124,21 +131,108 @@ def _message_value(signal): return msg.signals[signal.name]["value"] +def _scan_status_msg( + scan_name: str = "line_scan", + additional_scan_parameters: dict | None = None, + exp_time: float = 0.1, + frames_per_trigger: int = 2, + num_points: int = 3, +) -> ScanStatusMessage: + """Scan status message shaped like the one the v4 scan server publishes.""" + v4_info = ScanServerScanInfo( + scan_name=scan_name, + scan_id="test", + scan_type="software_triggered", + num_points=num_points, + exp_time=exp_time, + frames_per_trigger=frames_per_trigger, + additional_scan_parameters=additional_scan_parameters or {}, + ) + scan_parameters = {"exp_time": exp_time, "frames_per_trigger": frames_per_trigger} + info = v4_info.model_dump(mode="python") + info["scan_parameters"] = scan_parameters + return ScanStatusMessage( + scan_id="test", + status="open", + scan_name=scan_name, + scan_type="software_triggered", + num_points=num_points, + scan_parameters=scan_parameters, + info=info, + ) + + +TIMEPIX_SCAN_PARAMETERS = { + "time_roi_step": 1, + "time_roi_n": 4, + "idx_off_start": 0, + "idx_off_stop": 2, + "idx_on_start": 2, + "idx_on_stop": 4, +} + +LEVEL_SIGNALS = ( + "xes_level_1_off", + "xes_level_1_on", + "xes_level_2_off", + "xes_level_2_on", + "xes_diff_1_on_off", + "xes_diff_2_on_off", +) + + +def _set_scan_parameters(dev, scan_name: str, additional_scan_parameters: dict | None = None): + """Set the normalized scan parameters the way on_stage does, without staging.""" + dev.scan_info.msg = _scan_status_msg(scan_name, additional_scan_parameters) + dev.scan_parameters = fetch_scan_info(dev.scan_info) + + +def _xes_frames(spectra: np.ndarray): + """StartFrame, one XesData frame and EndFrame for a (TRoiN, NumEnergyPoints) array.""" + troin, n_energy_points = spectra.shape + start_frame = { + "type": "StartFrame", + "Mode": "TOA", + "TRoiStart": 0, + "TRoiStep": 1, + "TRoiN": troin, + "NumEnergyPoints": n_energy_points, + "save_interval": 1, + } + # TDSpectra is the flattened (time bin, energy point) array, energy points varying fastest + data_frames = [ + { + "type": "XesData", + "period": 131000, + "totalEvents": int(spectra.sum()), + "TDSpectra": spectra.ravel().tolist(), + "beforeROI": 0, + "afterROI": 0, + } + ] + end_frame = {"type": "EndFrame", "error": "", "periods": 4} + return start_frame, data_frames, end_frame + + +def _level_test_spectra() -> np.ndarray: + """4 time bins x 8 energy points; energy group 1 sums to [4, 8, 12, 16], group 2 to [40, 80, 120, 160].""" + time_bin = np.arange(1, 5, dtype=np.float32)[:, None] + return np.hstack([np.repeat(time_bin, 4, axis=1), np.repeat(10 * time_bin, 4, axis=1)]) + + @pytest.fixture(scope="function") def timepix(): """Timepix device with mocked EPICS signals and a fully mocked backend.""" backend = FakeBackend() - scan_info = SimpleNamespace( - msg=SimpleNamespace( - scan_name="step_scan", - scan_parameters={"exp_time": 0.1, "frames_per_trigger": 2}, - num_points=3, - scan_id="test", - ) - ) + scan_info = ScanInfo(msg=_scan_status_msg()) with ( mock.patch.object(ophyd, "cl") as mock_cl, mock.patch("superxas_bec.devices.timepix.timepix.TimepixFlyBackend", return_value=backend), + # The full default pixel map takes seconds to build; no test here depends on it. + mock.patch( + "superxas_bec.devices.timepix.timepix.create_single_energy_per_chip_pixel_map", + return_value=PixelMap(chips=[[{"i": 0, "p": 0, "f": [1.0]}]]), + ), ): mock_cl.get_pv = MockPV mock_cl.thread_class = threading.Thread @@ -379,6 +473,75 @@ def test_timepix_msg_buffer_callback_updates_xes_signals(timepix): assert _message_value(timepix.total_events) == 36 +def test_timepix_msg_buffer_callback_accumulates_xes_energy_per_scan(timepix): + """xes_accumulated_energy_1/2 sum the frames of one scan; staging starts a new sum.""" + timepix.troin = 4 + # one frame: xes_energy_1 = [4, 8, 12, 16], xes_energy_2 = [40, 80, 120, 160] + frames = _xes_frames(_level_test_spectra()) + + timepix.msg_buffer_callback(*frames) + timepix.msg_buffer_callback(*frames) + + np.testing.assert_array_equal( + _message_value(timepix.xes_data_accumulated_1), np.array([8, 16, 24, 32]) + ) + np.testing.assert_array_equal( + _message_value(timepix.xes_data_accumulated_2), np.array([80, 160, 240, 320]) + ) + assert timepix.xes_data_accumulated_1.get().metadata["async_update"]["type"] == "replace" + # the per-frame signal still carries only the last frame + np.testing.assert_array_equal(_message_value(timepix.xes_energy_1), np.array([4, 8, 12, 16])) + + _stage_for(timepix, "line_scan") + timepix.msg_buffer_callback(*frames) + + np.testing.assert_array_equal( + _message_value(timepix.xes_data_accumulated_1), np.array([4, 8, 12, 16]) + ) + + +@pytest.mark.parametrize("scan_name", [None, "line_scan"]) +def test_timepix_msg_buffer_callback_skips_levels_outside_timepix_scan(timepix, scan_name): + """Frames before the first stage or during other scans must not touch the on/off levels.""" + if scan_name is not None: + _set_scan_parameters(timepix, scan_name) + assert timepix._is_timepix_scan() is False + timepix.troin = 4 + with ExitStack() as stack: + level_puts = [ + stack.enter_context(mock.patch.object(getattr(timepix, name), "put")) + for name in LEVEL_SIGNALS + ] + timepix.msg_buffer_callback(*_xes_frames(_level_test_spectra())) + + for put in level_puts: + put.assert_not_called() + np.testing.assert_array_equal( + _message_value(timepix.xes_energy_1), np.array([4, 8, 12, 16], dtype=np.float32) + ) + + +def test_timepix_msg_buffer_callback_publishes_on_off_levels(timepix): + """During a timepix_scan the six level/diff signals are the means over the index windows.""" + _set_scan_parameters(timepix, "timepix_scan", TIMEPIX_SCAN_PARAMETERS) + timepix.troin = 4 + + timepix.msg_buffer_callback(*_xes_frames(_level_test_spectra())) + + # xes_energy_1 = [4, 8, 12, 16], xes_energy_2 = [40, 80, 120, 160] + # off window [0, 2), on window [2, 4) + expected = { + "xes_level_1_off": 6.0, + "xes_level_1_on": 14.0, + "xes_diff_1_on_off": 8.0, + "xes_level_2_off": 60.0, + "xes_level_2_on": 140.0, + "xes_diff_2_on_off": 80.0, + } + for name, value in expected.items(): + assert _message_value(getattr(timepix, name)) == pytest.approx(value), name + + def test_timepix_on_stop_stops_camera_writer_and_backend(timepix): """Stop should stop camera acquisition and delegate backend stop.""" timepix.cam.acquire.put(1) @@ -391,6 +554,182 @@ def test_timepix_on_stop_stops_camera_writer_and_backend(timepix): timepix.backend.on_stop.assert_called_once() +@pytest.mark.parametrize( + "window, expected", + [ + (slice(0, 2), 1.5), + (slice(1, 4), 3.0), + (slice(2, 2), np.nan), # empty + (slice(3, 2), np.nan), # start after stop + (slice(2, 5), np.nan), # stop past the end + (slice(4, 5), np.nan), # start past the end + (slice(-1, 2), np.nan), # negative start + ], +) +def test_window_mean(window, expected): + data = np.array([1.0, 2.0, 3.0, 4.0], dtype=np.float32) + np.testing.assert_equal(window_mean(data, window), expected) + + +def test_timepix_on_off_levels_are_nan_for_window_outside_received_bins(timepix): + """If TimePixFly sends fewer time bins than requested, the affected levels become NaN.""" + _set_scan_parameters(timepix, "timepix_scan", TIMEPIX_SCAN_PARAMETERS) + timepix.troin = 4 + # only 3 time bins: xes_energy_1 = [4, 8, 12], xes_energy_2 = [40, 80, 120] + frames = _xes_frames(_level_test_spectra()[:3]) + + with mock.patch("superxas_bec.devices.timepix.timepix.logger") as logger: + timepix.msg_buffer_callback(*frames) + timepix.msg_buffer_callback(*frames) + + assert _message_value(timepix.xes_level_1_off) == pytest.approx(6.0) + assert _message_value(timepix.xes_level_2_off) == pytest.approx(60.0) + for name in ("xes_level_1_on", "xes_level_2_on", "xes_diff_1_on_off", "xes_diff_2_on_off"): + assert np.isnan(_message_value(getattr(timepix, name))), name + # one warning per scan, not one per frame + assert logger.warning.call_count == 1 + assert "on [2, 4)" in logger.warning.call_args.args[0] + + +def _stage_for(dev, scan_name: str, additional_scan_parameters: dict | None = None, **kwargs): + """Stage the device for a scan, as the device server does (stage(), not on_stage()).""" + dev.scan_info.msg = _scan_status_msg(scan_name, additional_scan_parameters, **kwargs) + with mock.patch( + "superxas_bec.devices.timepix.timepix.get_full_path", + return_value="/tmp/timepix/test_scan.h5", + ): + return dev.stage() + + +def _prepare_settings_before_timepix_scan(dev): + """Typical beamline state: image writing on, XES off, default time binning.""" + _force_signal_value(dev.hdf.enable, "Enable") + dev.enable_xes = False + dev.troistep = 1 + dev.troin = 5000 + + +def _assert_settings_before_timepix_scan(dev): + assert dev.hdf.enable.get() == "Enable" + assert dev.enable_xes is False + assert dev.cam.tdc1_enable.get() == 0 + assert dev.troistep == 1 + assert dev.troin == 5000 + assert dev._settings_before_timepix_scan is None + + +def test_timepix_scan_stage_overrides_settings_and_unstage_restores_them(timepix, pixel_map): + """timepix_scan switches image writing off and XES on for the scan only.""" + timepix._pixel_map = pixel_map + _prepare_settings_before_timepix_scan(timepix) + params = {**TIMEPIX_SCAN_PARAMETERS, "time_roi_step": 2} + + with mock.patch.object(timepix.file_event, "put") as file_event_put: + _stage_for(timepix, "timepix_scan", params) + file_event_put.assert_not_called() + + # The device server only unstages devices that report Staged.yes + assert timepix._staged == Staged.yes + assert timepix.hdf.enable.get() != "Enable" + assert timepix.hdf.capture.get() != 1 + assert timepix.enable_xes is True + assert timepix.cam.tdc1_enable.get() == 1 + other_config = timepix.backend.on_stage.call_args.kwargs["other_config"] + assert (other_config.TRoiStep, other_config.TRoiN) == (2, 4) + + timepix.unstage() + + assert timepix._staged == Staged.no + _assert_settings_before_timepix_scan(timepix) + + +def test_timepix_unstage_after_other_scan_keeps_settings(timepix): + """Only timepix_scan touches image writing and XES.""" + _prepare_settings_before_timepix_scan(timepix) + + _stage_for(timepix, "line_scan") + timepix.unstage() + + _assert_settings_before_timepix_scan(timepix) + + +def test_timepix_stop_restores_settings_of_aborted_timepix_scan(timepix, pixel_map): + """An aborted scan is stopped but not unstaged; a later unstage must not restore again.""" + timepix._pixel_map = pixel_map + _prepare_settings_before_timepix_scan(timepix) + _stage_for(timepix, "timepix_scan", TIMEPIX_SCAN_PARAMETERS) + + timepix.stop() + + # XES was on during the scan, so the backend collection is stopped as well + timepix.backend.on_stop.assert_called_once() + _assert_settings_before_timepix_scan(timepix) + + # A user change after the abort survives the unstage the device server does before the + # next stage (no second restore). + timepix.set_enable_xes(True) + timepix.unstage() + assert timepix.enable_xes is True + + +def test_timepix_failed_restore_on_stop_is_retried_on_unstage(timepix, pixel_map): + timepix._pixel_map = pixel_map + _prepare_settings_before_timepix_scan(timepix) + _stage_for(timepix, "timepix_scan", TIMEPIX_SCAN_PARAMETERS) + working_set = timepix.hdf.enable.set + timepix.hdf.enable.set = mock.Mock(side_effect=TimeoutError("IOC did not answer")) + + timepix.stop() # logs the error instead of raising + + assert timepix._settings_before_timepix_scan is not None + timepix.hdf.enable.set = working_set + timepix.unstage() + _assert_settings_before_timepix_scan(timepix) + + +def test_timepix_failed_stage_restores_timepix_scan_settings(timepix, pixel_map): + """If on_stage raises after the overrides were applied, they are undone right away.""" + timepix._pixel_map = pixel_map + _prepare_settings_before_timepix_scan(timepix) + + with pytest.raises(ValueError, match="Exposure time"): + _stage_for(timepix, "timepix_scan", TIMEPIX_SCAN_PARAMETERS, exp_time=0.001) + + assert timepix._staged == Staged.no + _assert_settings_before_timepix_scan(timepix) + + +def test_timepix_failed_stage_and_failed_restore_keep_stage_error(timepix, pixel_map): + """A failing restore must not hide why staging failed; the next stage restores instead.""" + timepix._pixel_map = pixel_map + _prepare_settings_before_timepix_scan(timepix) + working_set = timepix.hdf.enable.set + + def set_or_time_out(value, **kwargs): + if value == "Enable": # only the restore writes the saved readback value + raise TimeoutError("IOC did not answer") + return working_set(value, **kwargs) + + timepix.hdf.enable.set = mock.Mock(side_effect=set_or_time_out) + + with mock.patch("superxas_bec.devices.timepix.timepix.logger") as logger: + with pytest.raises(ValueError, match="Exposure time"): + _stage_for(timepix, "timepix_scan", TIMEPIX_SCAN_PARAMETERS, exp_time=0.001) + + assert timepix._staged == Staged.no + assert timepix._settings_before_timepix_scan is not None + assert "IOC did not answer" in logger.error.call_args.args[0] + + # Not staged, so the device server will not unstage it: the next stage puts the settings + # back before staging for a scan that is not a timepix_scan. + timepix.hdf.enable.set = working_set + _stage_for(timepix, "line_scan") + + _assert_settings_before_timepix_scan(timepix) + # XES is off again, so TimePixFly is not configured for the line_scan + timepix.backend.on_stage.assert_not_called() + + def test_timepix_on_destroy_cleans_up_backend(timepix): """Destroy should stop polling and forward cleanup to the backend.""" timepix.on_destroy() @@ -425,3 +764,155 @@ def test_timepix_on_prescan_returns_correct_status(timepix): status.wait(timeout=1) assert status.done is True assert status.success is True + + +# --- Lost acquisitions: TimePixFly closes the frame before the exposure ends (S00437 point 1) --- + + +def _wait_for(condition, timeout: float = 2.0): + """Poll until condition() is true (the repetition runs in its own thread).""" + deadline = time.monotonic() + timeout + while not condition(): + assert time.monotonic() < deadline, "condition not met in time" + time.sleep(0.01) + + +def _camera_acquires(dev): + """The camera runs one acquisition: AcquireBusy DONE -> ACQUIRING -> DONE.""" + _force_signal_value(dev.cam.acquire_busy, ACQUIRESTATUS.ACQUIRING) + _force_signal_value(dev.cam.acquire_busy, ACQUIRESTATUS.DONE) + + +def _empty_frames(): + """Frames of an acquisition in which TimePixFly saw no TDC pulse (as S00437, point 1).""" + start_frame, data_frames, end_frame = _xes_frames(np.zeros((4, 8), dtype=np.float32)) + data_frames[0]["period"] = 0 + return start_frame, data_frames, {**end_frame, "periods": 0} + + +@pytest.fixture +def xes_timepix(timepix): + """Timepix armed for XES as in a timepix_scan: image writing off, 1 s exposure, 1 image.""" + _force_signal_value(timepix.hdf.enable, "Disable") + timepix.troin = 4 + timepix._exposure_per_trigger = 1.0 + timepix._images_per_trigger = 1 + backend_done = [StatusBase(), StatusBase()] + timepix.backend.on_trigger_finished = mock.Mock(side_effect=backend_done) + timepix.backend_done = backend_done + return timepix + + +def test_timepix_lost_acquisition_is_repeated_once(xes_timepix): + """A frame closed before the exposure ends is dropped and the acquisition repeated.""" + timepix = xes_timepix + with ( + mock.patch.object(timepix.xes_data, "put") as xes_put, + mock.patch.object(timepix.static_spectra, "put") as spectra_put, + # a 2 x 2 detector image + mock.patch.object(timepix.image.array_size.height, "get", return_value=2), + mock.patch.object(timepix.image.array_size.width, "get", return_value=2), + mock.patch.object(timepix.image.array_data, "get", return_value=np.arange(4)), + ): + status = timepix.on_trigger() + assert timepix.cam.acquire.get() == 1 + + # TimePixFly closes the frame at once, without TDC periods: nothing is published + timepix.msg_buffer_callback(*_empty_frames()) + xes_put.assert_not_called() + # the image of the lost acquisition is not published either + timepix._on_new_image_received(value=1, old_value=0) + spectra_put.assert_not_called() + + _camera_acquires(timepix) + timepix.backend_done[0].set_finished() + assert status.done is False + + # the repetition arms TimePixFly again and restarts the camera + _wait_for(lambda: timepix.backend.on_trigger_finished.call_count == 2) + timepix.msg_buffer_callback(*_xes_frames(_level_test_spectra())) + timepix._on_new_image_received(value=2, old_value=1) + _camera_acquires(timepix) + timepix.backend_done[1].set_finished() + status.wait(timeout=2) + + assert status.success is True + xes_put.assert_called_once() + np.testing.assert_array_equal(xes_put.call_args.args[0], _level_test_spectra()) + spectra_put.assert_called_once() + assert not timepix._xes_acquisitions + + +def test_timepix_lost_repetition_is_published(xes_timepix): + """If the repetition is lost too, its empty frame is published: the point keeps one row.""" + timepix = xes_timepix + with mock.patch.object(timepix.xes_data, "put") as xes_put: + status = timepix.on_trigger() + timepix.msg_buffer_callback(*_empty_frames()) + timepix._on_new_image_received(value=1, old_value=0) + _camera_acquires(timepix) + timepix.backend_done[0].set_finished() + _wait_for(lambda: timepix.backend.on_trigger_finished.call_count == 2) + + with mock.patch("superxas_bec.devices.timepix.timepix.logger") as logger: + timepix.msg_buffer_callback(*_empty_frames()) + _camera_acquires(timepix) + timepix.backend_done[1].set_finished() + status.wait(timeout=2) + + assert status.success is True + xes_put.assert_called_once() + assert "cannot be repeated" in logger.error.call_args.args[0] + + +def test_timepix_frame_without_tdc_after_exposure_is_published(xes_timepix): + """No TDC pulse during a whole exposure is a detector/TDC problem: publish, do not repeat.""" + timepix = xes_timepix + timepix._exposure_per_trigger = 0.0 # the exposure has ended when the frame arrives + with ( + mock.patch.object(timepix.xes_data, "put") as xes_put, + mock.patch("superxas_bec.devices.timepix.timepix.logger") as logger, + ): + status = timepix.on_trigger() + timepix.msg_buffer_callback(*_empty_frames()) + _camera_acquires(timepix) + timepix.backend_done[0].set_finished() + status.wait(timeout=1) + + assert status.success is True + xes_put.assert_called_once() + assert timepix.backend.on_trigger_finished.call_count == 1 + assert "no TDC pulse" in logger.error.call_args.args[0] + + +def test_timepix_lost_acquisition_not_repeated_while_writing_images(xes_timepix): + """With HDF5 image writing on, a repetition would add an image: keep the point as it is.""" + timepix = xes_timepix + _force_signal_value(timepix.hdf.enable, "Enable") + with mock.patch.object(timepix.xes_data, "put") as xes_put: + status = timepix.on_trigger() + timepix.msg_buffer_callback(*_empty_frames()) + _camera_acquires(timepix) + timepix.backend_done[0].set_finished() + status.wait(timeout=1) + + assert status.success is True + xes_put.assert_called_once() + assert timepix.backend.on_trigger_finished.call_count == 1 + + +def test_timepix_stop_cancels_repetition_of_lost_acquisition(xes_timepix): + """A stop while the lost acquisition is still running cancels the trigger, no repetition.""" + timepix = xes_timepix + status = timepix.on_trigger() + timepix.msg_buffer_callback(*_empty_frames()) + + timepix.stop() + + with pytest.raises(DeviceStoppedError): + status.wait(timeout=1) + _camera_acquires(timepix) + time.sleep(0.2) + assert timepix.backend.on_trigger_finished.call_count == 1 + assert not timepix._xes_acquisitions + assert timepix._images_to_skip == 0 diff --git a/tests/tests_scans/test_timepix_scan.py b/tests/tests_scans/test_timepix_scan.py new file mode 100644 index 0000000..87c4dce --- /dev/null +++ b/tests/tests_scans/test_timepix_scan.py @@ -0,0 +1,75 @@ +# pylint: skip-file +import pytest +from bec_lib.bec_errors import ScanInputValidationError +from bec_server.scan_server.tests.scan_fixtures import * + +from superxas_bec.scans.timepix_scan import TimepixScan + +TIMEPIX_PARAMETERS = { + "time_roi_step": 1, + "time_roi_n": 10, + "idx_off_start": 0, + "idx_off_stop": 4, + "idx_on_start": 5, + "idx_on_stop": 10, +} + + +def _assemble_timepix_scan(scan_assembler, **overrides): + params = {**TIMEPIX_PARAMETERS, **overrides} + return scan_assembler( + "timepix_scan", "samx", -1.0, 1.0, steps=3, relative=False, exp_time=0.1, **params + ) + + +def test_timepix_scan_forwards_parameters_to_scan_info(scan_assembler): + """The Timepix device reads the parameters from additional_scan_parameters.""" + scan = _assemble_timepix_scan(scan_assembler) + + assert isinstance(scan, TimepixScan) + assert scan.scan_info.scan_name == "timepix_scan" + for name, value in TIMEPIX_PARAMETERS.items(): + assert scan.scan_info.additional_scan_parameters[name] == value + + +@pytest.mark.parametrize( + "overrides, match", + [ + ({"idx_off_start": 5, "idx_off_stop": 4}, "idx_off_start"), + ({"idx_on_start": 9, "idx_on_stop": 6}, "idx_on_start"), + # start == stop would be an empty window + ({"idx_off_start": 3, "idx_off_stop": 3}, "idx_off_start"), + ({"idx_on_start": 7, "idx_on_stop": 7}, "idx_on_start"), + ], +) +def test_timepix_scan_rejects_start_not_smaller_than_stop(scan_assembler, overrides, match): + with pytest.raises(ValueError, match=match): + _assemble_timepix_scan(scan_assembler, **overrides) + + +@pytest.mark.parametrize( + "overrides, name", + [ + ({"idx_off_stop": 11}, "idx_off_stop"), + ({"idx_on_stop": 11}, "idx_on_stop"), + ({"idx_on_start": 11, "idx_on_stop": 12}, "idx_on_start"), + ], +) +def test_timepix_scan_rejects_index_above_time_roi_n(scan_assembler, overrides, name): + with pytest.raises(ValueError, match=f"{name} with value"): + _assemble_timepix_scan(scan_assembler, **overrides) + + +def test_timepix_scan_accepts_stop_equal_to_time_roi_n(scan_assembler): + """stop is exclusive, so a window may end at time_roi_n.""" + scan = _assemble_timepix_scan(scan_assembler, idx_on_start=9, idx_on_stop=10) + + assert scan.scan_info.additional_scan_parameters["idx_on_stop"] == 10 + + +@pytest.mark.parametrize( + "overrides", [{"idx_off_start": -1}, {"time_roi_n": 0}, {"time_roi_step": 0}] +) +def test_timepix_scan_argument_bounds_are_validated(scan_assembler, overrides): + with pytest.raises(ScanInputValidationError): + _assemble_timepix_scan(scan_assembler, **overrides)