fix(timepix): harden timepix_scan and repeat acquisitions with lost frames

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

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,

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

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)