diff --git a/superxas_bec/devices/timepix/timepix.py b/superxas_bec/devices/timepix/timepix.py index aa55a9f..2300747 100644 --- a/superxas_bec/devices/timepix/timepix.py +++ b/superxas_bec/devices/timepix/timepix.py @@ -12,6 +12,9 @@ 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 @@ -20,6 +23,9 @@ 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 Staged +from ophyd.status import StatusTimeoutError, WaitTimeoutError +from ophyd.utils import InvalidState from ophyd_devices import ( AsyncSignal, CompareStatus, @@ -166,6 +172,18 @@ 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 [start, stop), or NaN if the window is empty or out of range.""" + 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.""" @@ -193,6 +211,31 @@ DETECTOR_SHAPE = (512, 1024) # Shape of the TimePix detector TIMEPIX_FLY_BACKEND_DEFAULT_REDIS_TOPIC = "public/timepix_fly_backend/raw_data" +TIMEPIX_SCAN_NAME = "timepix_scan" + +# max wait for the images of a lost acquisition before repeating it +LOST_ACQUISITION_IMAGE_TIMEOUT = 2.0 + + +@dataclass +class XesAcquisition: + """ + One armed TimePixFly acquisition; EndFrames arrive in order, one per acquisition. + + Attributes: + exposure (float): exposure of this trigger in seconds. + repeat_allowed (bool): may be repeated if its raw data are lost. + acquire_time (float | None): time.monotonic() when Acquire was set. + lost (bool): frame closed before the exposure could end. + done (bool): acquisition status has completed. + """ + + exposure: float + repeat_allowed: bool + acquire_time: float | None = None + lost: bool = False + done: bool = False + class Timepix(PSIDeviceBase, TimePixControl): """ @@ -299,13 +342,37 @@ class Timepix(PSIDeviceBase, TimePixControl): self.accumulated_data_e1 = None self.accumulated_data_e2 = None + # settings saved before timepix_scan overrides, None when none are active + self._settings_before_timepix_scan: dict[str, Any] | None = None + self._on_off_window_warned = False + # armed acquisitions awaiting their EndFrame, oldest first + self._xes_acquisitions: deque[XesAcquisition] = deque() + self._xes_acquisitions_lock = threading.Lock() + # images of a lost acquisition still to arrive, 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 (it stages the AD plugins). Still marks Staged.yes, since + BEC only unstages staged devices. + """ 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: + # undo overrides; a failed restore is only logged so the stage error is raised + 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 [] @@ -366,6 +433,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 # will be repeated n_energy_points = start_frame.get("NumEnergyPoints", None) troin = start_frame["TRoiN"] if troin != self._troin: @@ -436,27 +505,106 @@ class Timepix(PSIDeviceBase, TimePixControl): ) ### Additional signals for timepix scan - if self.scan_parameters.scan_name == "timepix_scan": - idx_off_start = self.scan_parameters.additional_scan_parameters["idx_off_start"] - idx_off_stop = self.scan_parameters.additional_scan_parameters["idx_off_stop"] - idx_on_start = self.scan_parameters.additional_scan_parameters["idx_on_start"] - idx_on_stop = self.scan_parameters.additional_scan_parameters["idx_on_stop"] - level_1_off = np.mean(data_1[idx_off_start:idx_off_stop]) - level_1_on = np.mean(data_1[idx_on_start:idx_on_stop]) - level_2_off = np.mean(data_2[idx_off_start:idx_off_stop]) - level_2_on = np.mean(data_2[idx_on_start:idx_on_stop]) - 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) + 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 with zero TDC periods that closes before the exposure could end means the raw + data never reached TimePixFly (seen in S00437); it is repeated once if allowed. + + Args: + end_frame (dict): EndFrame from TimePixFly. + + Returns: + bool: True if the acquisition is lost and will be repeated. + """ + with self._xes_acquisitions_lock: + if not self._xes_acquisitions: + return False # nothing armed, 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 and on-off differences for both energy groups. + + Windows that do not fit the received time bins give NaN, keeping signals aligned with + the scan points. + + Args: + data_1 (np.ndarray): xes_energy_1 per time bin. + data_2 (np.ndarray): xes_energy_2 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 staged for a timepix_scan.""" + 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: @@ -632,6 +780,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 a lost acquisition; the repeat replaces it + self._images_to_skip -= 1 + return try: # Get new image data width = self.image.array_size.width.get() @@ -652,14 +805,15 @@ class Timepix(PSIDeviceBase, TimePixControl): """Called while staging the device.""" self.scan_parameters = fetch_scan_info(self.scan_info) - if self.scan_parameters.scan_name == "timepix_scan": - self._troistep = self.scan_parameters.additional_scan_parameters["time_roi_step"] - self._troin = self.scan_parameters.additional_scan_parameters["time_roi_n"] - self.hdf.enable.put(0) - self.set_enable_xes(True) + if self._is_timepix_scan(): + self._apply_timepix_scan_settings(self.scan_parameters.additional_scan_parameters) + else: + # leftover overrides from a failed restore + 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: @@ -668,6 +822,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) @@ -716,15 +873,48 @@ class Timepix(PSIDeviceBase, TimePixControl): timeout=self._pv_timeout ) + def _apply_timepix_scan_settings(self, params: dict) -> None: + """ + Apply timepix_scan overrides (time binning, HDF5 off, XES on), saving previous values. + + Args: + params (dict): additional_scan_parameters of the timepix_scan. + """ + if self._settings_before_timepix_scan is None: + # keep the original values if an earlier restore failed + 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"] + # waits for the readback; on_stage checks hdf.enable below + self.set_enable_image_writing(False) + self.set_enable_xes(True) + + def _restore_settings_after_timepix_scan(self) -> None: + """Restore settings saved by _apply_timepix_scan_settings; no-op if none are saved.""" + 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 on success so a failed restore is retried + 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) - if self.scan_parameters.scan_name == "timepix_scan": - self.hdf.enable.put(1) - self.set_enable_xes(False) + 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.""" @@ -744,51 +934,151 @@ 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, a lost acquisition is repeated once, but only with HDF5 writing off (a + repeat would add an extra image to the file). + """ + 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.""" + 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. + + Args: + point_status (DeviceStatus): trigger status returned by on_trigger. + repeat_allowed (bool): may be repeated if its raw data are lost. + + Raises: + TimeoutError: if TimePixFly is not ready in time. + """ + 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) + # register before arming: a lost frame's EndFrame can arrive immediately + 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.""" @@ -852,6 +1142,15 @@ class Timepix(PSIDeviceBase, TimePixControl): # Backend if self.enable_xes is True: self.backend.on_stop() + self._clear_xes_acquisitions() + # BEC does not unstage on abort; must run after backend stop (uses 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/timepix_scan.py b/superxas_bec/scans/timepix_scan.py index 802442f..bc4e6ee 100644 --- a/superxas_bec/scans/timepix_scan.py +++ b/superxas_bec/scans/timepix_scan.py @@ -47,20 +47,21 @@ class TimepixScan(LineScan): # 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", ge=0)], - idx_off_stop: Annotated[int, ScanArgument(display_name="Index Off Stop", description="Stop Index for Off Level", ge=0)], - idx_on_start: Annotated[int, ScanArgument(display_name="Index On Start", description="Start Index for On Level", ge=0)], - idx_on_stop: Annotated[int, ScanArgument(display_name="Index On Stop", description="Stop Index for On Level", ge=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, ): - if idx_off_start > idx_off_stop: + # windows are half-open [start, stop); empty ones would only give NaN + if idx_off_start >= idx_off_stop: raise ValueError( - f"idx_off_start ({idx_off_start}) cannot be bigger than idx_off_stop ({idx_off_stop})!" + f"idx_off_start ({idx_off_start}) must be smaller than idx_off_stop ({idx_off_stop})!" ) - if idx_on_start > idx_on_stop: + if idx_on_start >= idx_on_stop: raise ValueError( - f"idx_on_start ({idx_on_start}) cannot be bigger than idx_on_stop ({idx_on_stop})!" + 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), @@ -82,13 +83,6 @@ class TimepixScan(LineScan): **kwargs, ) - self.time_roi_step = time_roi_step - self.time_roi_n = time_roi_n - self.idx_off_start = idx_off_start - self.idx_off_stop = idx_off_stop - self.idx_on_start = idx_on_start - self.idx_on_stop = idx_on_stop - self.update_scan_info( time_roi_step=time_roi_step, time_roi_n=time_roi_n, diff --git a/tests/tests_devices/test_timepix.py b/tests/tests_devices/test_timepix.py index 0010025..c2c4164 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: + """v4-style scan status message.""" + 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 scan parameters as 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, + } + # energy points vary 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; groups sum to [4, 8, 12, 16] and [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 default pixel map is slow to build + 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): + """Accumulated signals sum frames per scan; staging resets them.""" + 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): + """On/off levels are untouched outside a timepix_scan.""" + 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): + """Level/diff signals are the window means during a timepix_scan.""" + _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,177 @@ 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): + """Too few time bins give NaN levels.""" + _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 + 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 via stage(), as the device server does.""" + 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): + """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 turns 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() + + 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): + """Stop restores settings; a later unstage does not restore again.""" + timepix._pixel_map = pixel_map + _prepare_settings_before_timepix_scan(timepix) + _stage_for(timepix, "timepix_scan", TIMEPIX_SCAN_PARAMETERS) + + timepix.stop() + + timepix.backend.on_stop.assert_called_once() + _assert_settings_before_timepix_scan(timepix) + + # a user change after the abort survives the next unstage + 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 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): + """A failing on_stage undoes the overrides.""" + 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 does not hide the stage error; the next stage restores.""" + 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 this + 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 next stage restores + timepix.hdf.enable.set = working_set + _stage_for(timepix, "line_scan") + + _assert_settings_before_timepix_scan(timepix) + 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 +759,153 @@ 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 + + +def _wait_for(condition, timeout: float = 2.0): + """Poll until condition() is true.""" + deadline = time.monotonic() + timeout + while not condition(): + assert time.monotonic() < deadline, "condition not met in time" + time.sleep(0.01) + + +def _camera_acquires(dev): + """Simulate one camera acquisition.""" + _force_signal_value(dev.cam.acquire_busy, ACQUIRESTATUS.ACQUIRING) + _force_signal_value(dev.cam.acquire_busy, ACQUIRESTATUS.DONE) + + +def _empty_frames(): + """Frames without any TDC period.""" + 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 set up for XES: 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): + """An early-closed frame 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 + + # frame closes at once: nothing is published + timepix.msg_buffer_callback(*_empty_frames()) + xes_put.assert_not_called() + 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 + + _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): + """A lost repetition is published as is.""" + 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 over a full exposure is published, not repeated.""" + timepix = xes_timepix + timepix._exposure_per_trigger = 0.0 # exposure already over + 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): + """No repetition with HDF5 image writing on.""" + 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 during a lost acquisition cancels the trigger without repeating.""" + 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..fe422c0 --- /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): + """Parameters are stored in 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"), + # 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)