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
+  - !include ./x10da_experimental_hutch.yaml

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."""

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

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,
+        )

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

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)