superxas_bec/device_configs/laser_delay_sim.yaml

@@ -0,0 +1,22 @@
+###################################
+##  Laser delay, simulated       ##
+###################################
+# SimLaserDelay: LaserDelay with soft signals, for a local demo BEC.
+#   bec.config.add_to_session("<superxas_bec>/superxas_bec/device_configs/laser_delay_sim.yaml")
+#   dev.laser_delay.calibrate(dev.laser_delay.sim_scope.get())
+# step_sleep is shortened for demos (beamline: 0.1); the calibration is kept in memory.
+
+laser_delay:
+  readoutPriority: baseline
+  description: Simulated laser delay in ns (no EPICS)
+  deviceClass: superxas_bec.devices.laser_delay_sim.SimLaserDelay
+  deviceConfig:
+    step_sleep: 0.01
+  deviceTags:
+    - laser
+    - timing
+    - simulation
+  onFailure: retry
+  enabled: true
+  readOnly: false
+  softwareTrigger: false

superxas_bec/device_configs/x10da_experimental_hutch.yaml

@@ -136,4 +136,11 @@ es2_try:
     prefix: X10DA-ES2-ET2:TRY
   onFailure: retry
   enabled: true
-  softwareTrigger: false
+  softwareTrigger: false
+
+###################################
+##   Laser delay (pump-probe)    ##
+###################################
+
+laser_delay_config:
+  - !include ./x10da_laser_delay.yaml

superxas_bec/device_configs/x10da_laser_delay.yaml

@@ -0,0 +1,33 @@
+###################################
+##   Laser delay (pump-probe)    ##
+###################################
+# Delay card + RF phase shifter 2 (ES4), Duetto laser. Docs: LaserDelay docstrings.
+# Calibrate after every laser restart: dev.laser_delay.calibrate(<scope value>); the calibration
+# is kept in the X10DA-LAS-DELAY: soft records, see dev.laser_delay.get_calibration_info().
+# onFailure buffer: an IOC outage does not abort unrelated scans.
+
+laser_delay:
+  readoutPriority: baseline
+  description: Laser delay in ns (delay card + phase shifter 2, Duetto); calibrate after laser restart
+  deviceClass: superxas_bec.devices.laser_delay.LaserDelay
+  deviceConfig:
+    delay_coarse_pv: X10DA-ES2-TIM:SET1-DELAY
+    delay_fine_pv: X10DA-ES2-TIM:SET1-FINE
+    evr_period_pv: X10DA-ES2-TIM:EVR-FPS2
+    phase_v1_pv: X10DA-ES4:AO-02
+    phase_v2_pv: X10DA-ES4:AO-01
+    rf_period: 2.0
+    cavity_period: 12.01
+    vmax: 1.0
+    steps_per_rf_period: 36
+    step_sleep: 0.1
+    phase_path: script  # "shortest" is opt-in
+    fine_rollover: true  # FINE 200 -> next coarse unit (open question)
+    calibration_pv_prefix: "X10DA-LAS-DELAY:"  # autosaved soft records with the calibration
+  deviceTags:
+    - laser
+    - timing
+  onFailure: buffer
+  enabled: true
+  readOnly: false
+  softwareTrigger: false

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/laser_delay_sim.py

@@ -0,0 +1,265 @@
+"""Simulated ``LaserDelay`` (no EPICS, simulated laser and scope) for demos and tests."""
+
+from __future__ import annotations
+
+import copy
+import inspect
+import math
+import threading
+from typing import Callable
+
+import numpy as np
+from bec_lib.logger import bec_logger
+from ophyd import Component as Cpt
+from ophyd import Kind, Signal
+
+from superxas_bec.devices.laser_delay import (
+    DELAY_CARD_FINE_PER_COARSE,
+    PHASE_VOLTAGE_LIMIT,
+    CheckedSignal,
+    ComputedPhaseSignal,
+    LaserDelay,
+    _DeviceWritesOnly,
+    delay_card_time,
+)
+
+logger = bec_logger.logger
+
+
+# per device name, for the lifetime of the process (survives a config reload)
+_SIM_CALIBRATIONS: dict[str, dict] = {}
+
+
+class MemoryCalibrationStore:
+    """In-memory stand-in for the calibration PVs (simulation and tests), keyed by device name."""
+
+    def __init__(self, name: str, registry: dict[str, dict] | None = None):
+        self.name = name
+        self.registry = _SIM_CALIBRATIONS if registry is None else registry
+
+    def __str__(self) -> str:
+        return f"memory ({self.name})"
+
+    @property
+    def content(self) -> dict | None:
+        """The stored content (for tests); None if nothing is stored."""
+        return self.registry.get(self.name)
+
+    @content.setter
+    def content(self, value: dict | None) -> None:
+        if value is None:
+            self.registry.pop(self.name, None)
+        else:
+            self.registry[self.name] = copy.deepcopy(value)
+
+    def load(self) -> dict | None:
+        """Stored content, or None."""
+        content = self.registry.get(self.name)
+        return copy.deepcopy(content) if content is not None else None
+
+    def save(self, content: dict) -> None:
+        """Replace the stored content."""
+        self.registry[self.name] = copy.deepcopy(content)
+
+    def destroy(self) -> None:
+        """Nothing to release; the content stays for the next device object."""
+
+
+class _HardwareSoftSignal(_DeviceWritesOnly, CheckedSignal):
+    """Soft stand-in for a hardware PV."""
+
+
+def _sim_range(name: str, low: float, high: float) -> Callable[[object], None]:
+    def check(value):
+        try:
+            number = float(value)
+        except (TypeError, ValueError) as exc:
+            raise ValueError(f"{name} must be a number, got {value!r}") from exc
+        if not low <= number <= high:
+            raise ValueError(f"{name} must be within [{low}, {high}], got {value!r}")
+
+    return check
+
+
+class SimLaserDelay(LaserDelay):
+    """LaserDelay with soft signals instead of PVs and a simulated laser and oscilloscope.
+
+    The ``*_pv`` arguments are ignored. Start: laser period 2.0002 µs, card 150 / 37, V1 = 1,
+    V2 = 0. The laser pulse train follows the phase shifter (unwrapped) and fires at or after the
+    delay card trigger. ``sim_scope`` (X-ray minus laser) starts at -0.88 ns, so
+    ``calibrate(dev.laser_delay.sim_scope.get())`` is a correct calibration; after a move it
+    equals the readback within 0.01 ns. ``sim_response`` is a noisy pump-probe step.
+    ``sim_set_hardware(...)`` changes the PVs as from outside BEC, to try the safety checks.
+
+    Args:
+        sim_cavity_period (float | None): simulated cavity period in ns; None: ``cavity_period``.
+        sim_time_zero (float): scope time of the pump-probe step in ns.
+        sim_rise (float): width of the step in ns.
+        sim_noise (float): noise standard deviation on ``sim_response``.
+    """
+
+    USER_ACCESS = LaserDelay.USER_ACCESS + ["sim_set_hardware"]
+    MIN_STEP_SLEEP = 0.0
+
+    delay_coarse = Cpt(
+        _HardwareSoftSignal,
+        value=150,
+        kind=Kind.normal,
+        validator=_sim_range("delay_coarse", 0, 65535),
+        doc="Delay card, units of 2 ns",
+    )
+    delay_fine = Cpt(
+        _HardwareSoftSignal,
+        value=37,
+        kind=Kind.normal,
+        validator=_sim_range("delay_fine", 0, DELAY_CARD_FINE_PER_COARSE - 1),
+        doc="Delay card, units of 0.01 ns",
+    )
+    evr_period = Cpt(Signal, value=2.0002, kind=Kind.config, doc="EVR laser period in µs")
+    phase_v1 = Cpt(
+        _HardwareSoftSignal,
+        value=1.0,
+        kind=Kind.normal,
+        validator=_sim_range("phase_v1", -PHASE_VOLTAGE_LIMIT, PHASE_VOLTAGE_LIMIT),
+        doc="Phase shifter V1",
+    )
+    phase_v2 = Cpt(
+        _HardwareSoftSignal,
+        value=0.0,
+        kind=Kind.normal,
+        validator=_sim_range("phase_v2", -PHASE_VOLTAGE_LIMIT, PHASE_VOLTAGE_LIMIT),
+        doc="Phase shifter V2",
+    )
+    sim_scope = Cpt(
+        ComputedPhaseSignal,
+        compute="read_sim_scope",
+        kind=Kind.normal,
+        doc="Simulated oscilloscope: X-ray minus laser in ns",
+    )
+    sim_response = Cpt(
+        ComputedPhaseSignal,
+        compute="read_sim_response",
+        kind=Kind.hinted,
+        doc="Simulated pump-probe signal",
+    )
+
+    # X-ray arrival and pulse train offset in ns; the start values give a scope reading of -0.88 ns
+    _SIM_XRAY_TIME = 305.495
+    _SIM_TRAIN_OFFSET = 6.125
+
+    def __init__(
+        self,
+        prefix: str = "",
+        *,
+        name: str,
+        sim_cavity_period: float | None = None,
+        sim_time_zero: float = 0.0,
+        sim_rise: float = 0.3,
+        sim_noise: float = 0.01,
+        **kwargs,
+    ):
+        super().__init__(prefix=prefix, name=name, **kwargs)
+        if self._store is None:  # no calibration PVs in the simulation: keep it in memory
+            self._store = MemoryCalibrationStore(self.name)
+        self._sim_cavity_period = sim_cavity_period
+        self._sim_time_zero = float(sim_time_zero)
+        self._sim_rise = float(sim_rise)
+        self._sim_noise = float(sim_noise)
+        self._sim_rng = np.random.default_rng()
+        self._sim_lock = threading.Lock()
+        self._sim_angle = math.atan2(self.phase_v2.get(), self.phase_v1.get())
+        self._sim_phase_turned = 0.0  # unwrapped phase shift in ns since the start
+        self.phase_v1.subscribe(self._sim_track_phase, run=False)
+        self.phase_v2.subscribe(self._sim_track_phase, run=False)
+
+    def sim_set_hardware(
+        self,
+        delay_coarse: int | None = None,
+        delay_fine: int | None = None,
+        phase_v1: float | None = None,
+        phase_v2: float | None = None,
+    ) -> None:
+        """Change the simulated PVs as from outside BEC (caput, IOC reboot)."""
+        with self._hardware_writes():
+            for key, value in (
+                ("delay_coarse", delay_coarse),
+                ("delay_fine", delay_fine),
+                ("phase_v1", phase_v1),
+                ("phase_v2", phase_v2),
+            ):
+                if value is not None:
+                    getattr(self, key).put(value)
+
+    def _sim_track_phase(self, *_, **__) -> None:
+        """Track the unwrapped phase (the laser's RF lock)."""
+        with self._sim_lock:
+            v1, v2 = self.phase_v1.get(), self.phase_v2.get()
+            if math.hypot(v1, v2) == 0:
+                return  # no phase defined; the lock keeps the last one
+            angle = math.atan2(v2, v1)
+            turned = (angle - self._sim_angle + math.pi) % (2 * math.pi) - math.pi
+            self._sim_angle = angle
+            self._sim_phase_turned += float(self.rf_period.get()) * turned / (2 * math.pi)
+
+    def on_connected(self) -> None:
+        """
+        Called after the device is connected and its signals are connected.
+        Default values for signals should be set here.
+        """
+        # soft signals lose their state on a rebuild; restore it before the calibration
+        try:
+            content = self._read_saved_calibration()
+            sim = (content or {}).get("sim")
+            if sim:
+                self.sim_set_hardware(**sim.get("hardware", {}))
+                with self._sim_lock:  # undo the turns tracked from the puts above
+                    self._sim_angle = math.atan2(self.phase_v2.get(), self.phase_v1.get())
+                    self._sim_phase_turned = float(sim["phase_turned"])
+        except Exception as exc:  # pylint: disable=broad-except
+            logger.error(f"{self.name}: could not restore the simulated hardware: {exc}")
+        super().on_connected()
+
+    def _extra_saved_state(self) -> dict:
+        hardware = {
+            "delay_coarse": int(self.delay_coarse.get()),
+            "delay_fine": int(self.delay_fine.get()),
+            "phase_v1": float(self.phase_v1.get()),
+            "phase_v2": float(self.phase_v2.get()),
+        }
+        with self._sim_lock:
+            return {"sim": {"phase_turned": self._sim_phase_turned, "hardware": hardware}}
+
+    def read_sim_scope(self) -> float:
+        """X-ray minus laser in ns, as the oscilloscope would show it."""
+        cavity = self._sim_cavity_period or float(self.cavity_period.get())
+        trigger = delay_card_time(self.delay_coarse.get(), self.delay_fine.get())
+        with self._sim_lock:
+            train = self._SIM_TRAIN_OFFSET + self._sim_phase_turned
+        laser = train + math.ceil((trigger - train) / cavity) * cavity
+        return self._SIM_XRAY_TIME - laser
+
+    def read_sim_response(self) -> float:
+        """Pump-probe signal: step at ``sim_time_zero`` (scope time) plus noise."""
+        x = (self.read_sim_scope() - self._sim_time_zero) / (math.sqrt(2) * self._sim_rise)
+        noise = self._sim_rng.normal(0, self._sim_noise) if self._sim_noise > 0 else 0.0
+        return 0.5 * (1 + math.erf(x)) + noise
+
+
+def _merged_signature(init: Callable, base_init: Callable) -> inspect.Signature:
+    """``init``'s parameters plus ``base_init``'s keyword ones (BEC only passes declared keys)."""
+    own = [p for p in inspect.signature(init).parameters.values() if p.kind != p.VAR_KEYWORD]
+    names = {p.name for p in own}
+    base = [
+        p
+        for p in inspect.signature(base_init).parameters.values()
+        if p.kind == p.KEYWORD_ONLY and p.name not in names
+    ]
+    var_kw = [
+        p for p in inspect.signature(base_init).parameters.values() if p.kind == p.VAR_KEYWORD
+    ]
+    return inspect.signature(init).replace(parameters=own + base + var_kw)
+
+
+SimLaserDelay.__init__.__signature__ = _merged_signature(
+    SimLaserDelay.__init__, LaserDelay.__init__
+)

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,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."""
 
@@ -192,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):
     """
@@ -221,9 +265,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 +312,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 +323,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 +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 []
@@ -355,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:
@@ -407,27 +487,124 @@ 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
+            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:
@@ -603,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()
@@ -621,9 +803,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:
+            # 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:
@@ -632,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)
@@ -680,12 +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)
+        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 +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."""
@@ -813,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/__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,93 @@
+"""
+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,
+    ):
+        # 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}) 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/reference/DelayLaser.py.txt

@@ -0,0 +1,234 @@
+#!/usr/bin/python
+# Reference only: the beamline scientist's laser-delay script, attached to request 2.
+# Kept verbatim (Python 2 era, CaChannel/epicsPV) as the behavioural spec for the BEC
+# LaserDelay device. Do not import or run it from BEC code.
+
+from CaChannel import ca, CaChannel
+from epicsPV import epicsPV
+from math import *
+from numpy import *
+import time
+
+#Possible values of Laser: Duetto, KML, XITON, PI
+Laser="Duetto"
+Beamline="SuperXAS"
+#LaserTriggerMode: "internal" or "external" External means from Event receiver.
+
+
+LaserTriggerMode="external"
+
+# Units of time are nanoseconds
+
+#If internal triggering is used, set LaserPeriod (time between two laser pulses) as it is set at the laser
+LaserPeriod=0
+if (LaserTriggerMode=="internal"): LaserPeriod=2000.2
+
+
+#Timing mode "DelayAndPhase" - use both Delay card and phase shifter, "Delay" - use only delay card, "Phase" - use only phase shifter
+if (Laser=="Duetto"):
+     TimingMode="DelayAndPhase"
+
+#Index Of delay channel (as in EPICS name) used for laser
+DelayChIndex=1
+EVRChIndex=2
+
+#For SuperXAS 1 corresponds to the old (KML) phase shifter
+#For SuperXAS 2 corresponds to ES4 phase shifter
+PhaseShifterIndex=2
+
+
+
+#change only for another synchrotron
+RFPeriod=2.0
+
+DelayCh=CaChannel()
+DelayChFine=CaChannel()
+EVRCh=CaChannel()
+PhaseV1=CaChannel()
+PhaseV2=CaChannel()
+
+
+def ConnectEpics():
+
+	global CavityPeriod, Laser, Beamline
+	global DelayCh, DelayChFine, EVRCh, PhaseV1, PhaseV2
+
+	if (Laser=="Duetto"):
+		CavityPeriod=12.01
+
+	if (Beamline=="SuperXAS"):
+		DN="X10DA-ES2-TIM:SET"+str(DelayChIndex)
+		EVRChName="X10DA-ES2-TIM:EVR-FPS"+str(EVRChIndex)
+		if (PhaseShifterIndex==1):
+			PhaseV1Name="X10DA-LAS-PHSH:V1"
+			PhaseV2Name="X10DA-LAS-PHSH:V2"
+		if (PhaseShifterIndex==2):
+			PhaseV1Name="X10DA-ES4:AO-02"
+			PhaseV2Name="X10DA-ES4:AO-01"
+
+	if (Beamline=="Phoenix"):
+		DN="X07MB-ES-TIM:SET"+str(DelayChIndex)
+		EVRChName="X07MB-ES-TIM:EVR-FPS"+str(EVRChIndex)
+		PhaseV1Name="X07MB-ES1-PP2:VO2"
+		PhaseV2Name="X07MB-ES1-PP2:VO1"
+
+	DelayChName=DN+"-DELAY"
+	DelayChFineName=DN+"-FINE"
+
+
+
+
+
+
+
+
+	if ((TimingMode=="Delay") or (TimingMode=="DelayAndPhase")):
+		DelayCh.searchw(DelayChName)
+		DelayChFine.searchw(DelayChFineName)
+
+	if ((TimingMode=="Phase") or (TimingMode=="DelayAndPhase")):
+		PhaseV1.searchw(PhaseV1Name)
+		PhaseV2.searchw(PhaseV2Name)
+
+	if (LaserTriggerMode=="external"):
+		EVRCh.searchw(EVRChName)
+
+#Maybe will be useful to automatically set pulse width for different lasers
+#def SetPulseWidth():
+
+
+def CurrentSituation(FromScopeXrayMinusLaser):
+
+	global LaserPeriod, OffsetT
+	if (TimingMode=="Delay"):
+		PhaseT=0
+	if (TimingMode=="Phase"):
+		DelayCardT=0
+
+	if (LaserTriggerMode=="external"):
+	    LaserPeriod=1000*EVRCh.getw()
+	    print ("LaserPeriod=", LaserPeriod)
+
+	if ((TimingMode=="Delay") or (TimingMode=="DelayAndPhase")):
+		dt1=DelayCh.getw()
+		dt2=DelayChFine.getw()
+		DelayCardT=2.0*dt1+0.01*dt2
+
+
+
+	if ((TimingMode=="Phase") or (TimingMode=="DelayAndPhase")):
+		PhaseT=GetPhaseT()
+		Ph=FromScopeXrayMinusLaser%CavityPeriod
+		OffsetPhaseT=-PhaseT-Ph
+	OffsetT=-DelayCardT-FromScopeXrayMinusLaser
+
+	print ("LaserPeriod=", LaserPeriod)
+	print ("OffsetT=", OffsetT)
+	print ("Current PhaseT", PhaseT)
+	print ("Offset Phse shifter T", OffsetPhaseT)
+	qwe=(LaserPeriod, OffsetT, OffsetPhaseT, PhaseT)
+# 	print ("qwe=",qwe)
+	savetxt("Timing.offset", qwe)
+
+
+
+def GetPhaseT():
+	v1=PhaseV1.getw()
+	v2=PhaseV2.getw()
+	phi=atan2(v2,v1)
+	T=RFPeriod*phi/(2*pi)
+	return T
+
+
+
+def SetVoltages(phi):
+
+	#Vmax may be different for different lasers or phase shifters
+	#if (PhaseShifterIndex==1):
+	Vmax=1.0
+
+	V1=Vmax*cos(phi)
+	V2=Vmax*sin(phi)
+	PhaseV1.putw(V1)
+	PhaseV2.putw(V2)
+	#set values via epics
+	print (V1, V2)
+	time.sleep(0.1)
+
+
+
+
+def MovePhaseShifter(RelativeDelay):
+
+
+
+	NumStepsPerPeriod=36
+	DelayStep=sign(RelativeDelay)*RFPeriod/NumStepsPerPeriod
+
+	#read current values of
+
+	CurT=GetPhaseT()
+	for t1 in arange(0,RelativeDelay, DelayStep):
+		phi1=2*pi*(CurT+t1)/RFPeriod
+		SetVoltages(phi1)
+
+	phi1=2*pi*(CurT+RelativeDelay)/RFPeriod
+	SetVoltages(phi1)
+
+
+
+
+def MoveDelayCard(NewDelayTime):
+	print ("ndt,lp", NewDelayTime, LaserPeriod)
+	t1=NewDelayTime%LaserPeriod
+	t2=int(t1/2.0)
+	t3=int((t1%2.0)/0.009999)
+	print ("t1 t2 t3 ",t1,t2,t3)
+	DelayCh.putw(t2)
+	DelayChFine.putw(t3)
+
+
+
+def SetDelay(t=0):
+	global LaserPeriod, OffsetT
+	a=loadtxt("Timing.offset")
+	LaserPeriod=a[0]
+	OffsetT=a[1]
+	print ("a1=",a[1])
+	OffsetPhaseT=a[2]
+	print ("OffsetPhaseT from File", OffsetPhaseT)
+	CurrentPhaseT=a[3]
+	print ("Current value of PhaseT", CurrentPhaseT)
+
+	TotalT=-OffsetT-t
+	ForPhaseShifter=-t%CavityPeriod-OffsetPhaseT
+	print ("-t/Cavity", -t%CavityPeriod, "-OffsetPhaseT", OffsetPhaseT, "-CurrentPhaseT", CurrentPhaseT)
+	MoveDelayCard(TotalT)
+	print ("Relative Phase shifter delay", ForPhaseShifter-CurrentPhaseT)
+	MovePhaseShifter(ForPhaseShifter-CurrentPhaseT)
+	print ("OffsetT before save=", OffsetT)
+	savetxt("Timing.offset",(LaserPeriod,OffsetT,OffsetPhaseT,ForPhaseShifter))
+
+
+def main():
+
+
+        ConnectEpics()
+#        CurSit=-20.015+15.46
+#	#CurSit=-14.58+15.46
+#	CurSit=546.04-546.92
+        #CurSit=0.1
+        #CurrentSituation(FromScopeXrayMinusLaser=CurSit)
+
+        print ("------------------")
+
+        SetDelay(-4.9)
+
+#	SetDelay(0.5)
+#	time.sleep(5)
+#	SetDelay(.5)
+if __name__=="__main__":
+	main()
+#MovePhaseShifter(2.0)
+#MoveDelayCard(25.)
+#time.sleep(10)

tests/tests_devices/test_laser_delay_golden.py

@@ -0,0 +1,427 @@
+"""Golden tests: the device writes exactly what the reference ``DelayLaser.py`` writes."""
+
+# pylint: disable=redefined-outer-name,protected-access,no-member
+
+from __future__ import annotations
+
+import math
+import sys
+import types
+from dataclasses import dataclass, field
+from pathlib import Path
+
+import numpy as np
+import pytest
+from ophyd_devices.tests.utils import patched_device
+
+from superxas_bec.devices.laser_delay import (
+    LaserDelay,
+    LaserDelayHardwareError,
+    delay_card_setting,
+    phase_from_voltages,
+    phase_ramp_offsets,
+    voltages_for_phase,
+    wrap_phase,
+)
+
+REFERENCE_SCRIPT = Path(__file__).parent / "reference" / "DelayLaser.py.txt"
+
+DELAY = "X10DA-ES2-TIM:SET1-DELAY"
+FINE = "X10DA-ES2-TIM:SET1-FINE"
+EVR = "X10DA-ES2-TIM:EVR-FPS2"
+V1 = "X10DA-ES4:AO-02"
+V2 = "X10DA-ES4:AO-01"
+
+FLOAT_TOL = 1e-12
+
+
+@pytest.fixture(autouse=True)
+def fast_ramp(monkeypatch):
+    """Tests ramp without sleeping; the beamline minimum is tested on its own."""
+    monkeypatch.setattr(LaserDelay, "MIN_STEP_SLEEP", 0.0)
+
+
+# start states
+STATE_A = {EVR: 2.0002, DELAY: 150, FINE: 37, V1: 0.6, V2: -0.8}
+STATE_B = {EVR: 1.0, DELAY: 10, FINE: 150, V1: -0.3, V2: 0.95}
+
+
+@dataclass
+class Scenario:
+    """Calibrate with ``scope``, then run ``steps`` (delays, or dicts of PVs changed by hand)."""
+
+    initial: dict
+    scope: float
+    steps: list = field(default_factory=list)
+
+
+SCENARIOS = {
+    "scope_cal_and_basic_moves": Scenario(STATE_A, 546.04 - 546.92, [-4.9, 0.0, 0.5, 5.3]),
+    "cross_cavity_period": Scenario(STATE_A, 546.04 - 546.92, [11.9, 12.2, 11.9]),
+    "large_negative_delays": Scenario(STATE_A, 546.04 - 546.92, [-1500.3, -250.37, -24.02]),
+    "laser_period_wrap": Scenario(STATE_A, 546.04 - 546.92, [400.0, -1800.0, 299.49, 299.5]),
+    "large_phase_moves_both_directions": Scenario(
+        STATE_A, 546.04 - 546.92, [0.1, 6.0, 0.1, 11.95, 0.05]
+    ),
+    "other_state_positive_scope": Scenario(STATE_B, 37.25, [37.25, 30.0, -3.33, 1000.0]),
+    "other_state_wrap_short_period": Scenario(STATE_B, -14.58 + 15.46, [-990.0, 2.5, 999.99]),
+}
+
+
+# reference script with fake EPICS
+
+
+class FakeIOC:
+    """PV values shared by all fake channels; records puts in order."""
+
+    def __init__(self, initial: dict):
+        self.values = dict(initial)
+        self.puts: list[tuple[str, object]] = []
+
+
+def load_reference_script(ioc: FakeIOC, monkeypatch, tmp_path, sleeps: list) -> types.ModuleType:
+    """Execute the reference script as a fresh module talking to ``ioc``."""
+
+    class CaChannel:  # pylint: disable=invalid-name
+        """Fake of CaChannel.CaChannel: searchw/getw/putw."""
+
+        def __init__(self):
+            self.pvname = None
+
+        def searchw(self, pvname):
+            self.pvname = pvname
+
+        def getw(self):
+            return ioc.values[self.pvname]
+
+        def putw(self, value):
+            ioc.puts.append((self.pvname, value))
+            ioc.values[self.pvname] = value
+
+    ca_module = types.ModuleType("CaChannel")
+    ca_module.CaChannel = CaChannel
+    ca_module.ca = types.SimpleNamespace()
+    epics_module = types.ModuleType("epicsPV")
+    epics_module.epicsPV = object
+    monkeypatch.setitem(sys.modules, "CaChannel", ca_module)
+    monkeypatch.setitem(sys.modules, "epicsPV", epics_module)
+    monkeypatch.chdir(tmp_path)  # Timing.offset is written to the working directory
+
+    module = types.ModuleType("delay_laser_reference")
+    module.__file__ = str(REFERENCE_SCRIPT)
+    source = REFERENCE_SCRIPT.read_text(encoding="utf-8")
+    code = compile(source, str(REFERENCE_SCRIPT), "exec")
+    exec(code, module.__dict__)  # pylint: disable=exec-used
+    module.time = types.SimpleNamespace(sleep=sleeps.append)
+    module.ConnectEpics()
+    return module
+
+
+def run_reference(scenario: Scenario, monkeypatch, tmp_path) -> dict:
+    """Run a scenario through the script. Returns calibration, puts and state per move."""
+    ioc = FakeIOC(scenario.initial)
+    sleeps: list = []
+    script = load_reference_script(ioc, monkeypatch, tmp_path, sleeps)
+    script.CurrentSituation(scenario.scope)
+    result = {"calibration": list(np.loadtxt("Timing.offset")), "moves": []}
+    for step in scenario.steps:
+        if isinstance(step, dict):
+            ioc.values.update(step)
+            continue
+        ioc.puts.clear()
+        sleeps.clear()
+        script.SetDelay(step)
+        result["moves"].append(
+            {
+                "target": step,
+                "puts": list(ioc.puts),
+                "sleeps": len(sleeps),
+                "state": list(np.loadtxt("Timing.offset")),
+            }
+        )
+    return result
+
+
+# device with recorded puts
+
+
+def record_puts(device: LaserDelay) -> list:
+    """Record every put to the four written PVs of ``device``, in order."""
+    log: list[tuple[str, object]] = []
+    for signal in (device.delay_coarse, device.delay_fine, device.phase_v1, device.phase_v2):
+        pv = signal._write_pv
+        original = pv.put
+
+        def recorded(value, *args, _pv=pv, _original=original, **kwargs):
+            log.append((_pv.pvname, value))
+            return _original(value, *args, **kwargs)
+
+        pv.put = recorded
+    return log
+
+
+def set_pv_values(device: LaserDelay, values: dict) -> None:
+    """Set mocked PV values without recording them as puts."""
+    signals = {
+        DELAY: device.delay_coarse,
+        FINE: device.delay_fine,
+        EVR: device.evr_period,
+        V1: device.phase_v1,
+        V2: device.phase_v2,
+    }
+    for pvname, value in values.items():
+        signals[pvname]._read_pv.mock_data = value
+
+
+def run_device(scenario: Scenario, **device_kwargs) -> dict:
+    """Run a scenario through the device (default configuration unless ``device_kwargs``)."""
+    device_kwargs.setdefault("fine_rollover", False)  # the script writes FINE 200
+    with patched_device(LaserDelay, name="laser_delay", step_sleep=0.0, **device_kwargs) as device:
+        set_pv_values(device, scenario.initial)
+        log = record_puts(device)
+        calibration = device.calibrate(scenario.scope)
+        result = {
+            "calibration": [
+                calibration[key]
+                for key in ("laser_period", "offset_t", "offset_phase_t", "phase_t")
+            ],
+            "moves": [],
+        }
+        for step in scenario.steps:
+            if isinstance(step, dict):
+                set_pv_values(device, step)
+                continue
+            log.clear()
+            status = device.set(step)
+            status.wait(timeout=30)
+            assert status.success
+            state = device.get_calibration()
+            result["moves"].append(
+                {
+                    "target": step,
+                    "puts": list(log),
+                    "state": [
+                        state[key]
+                        for key in ("laser_period", "offset_t", "offset_phase_t", "phase_t")
+                    ],
+                    "readback": device.read()["laser_delay"]["value"],
+                }
+            )
+        return result
+
+
+# comparison helpers
+
+
+def assert_same_value(expected, actual, context: str) -> None:
+    """Integers must match exactly (and be integers); floats within FLOAT_TOL."""
+    if isinstance(expected, (int, np.integer)) and not isinstance(expected, bool):
+        assert isinstance(actual, (int, np.integer)), f"{context}: {actual!r} is not an integer"
+        assert int(actual) == int(expected), f"{context}: {actual} != {expected}"
+        return
+    assert math.isclose(
+        float(actual), float(expected), rel_tol=FLOAT_TOL, abs_tol=FLOAT_TOL
+    ), f"{context}: {actual!r} != {expected!r} (diff {float(actual) - float(expected):.3e})"
+
+
+def assert_same_puts(expected: list, actual: list, context: str) -> None:
+    """Same PVs in the same order, same values."""
+    assert [pv for pv, _ in actual] == [
+        pv for pv, _ in expected
+    ], f"{context}: PV sequence differs (expected {len(expected)} puts, got {len(actual)})"
+    for index, ((pv, expected_value), (_, actual_value)) in enumerate(zip(expected, actual)):
+        assert_same_value(expected_value, actual_value, f"{context}, put #{index} to {pv}")
+
+
+# tests
+
+
+@pytest.mark.parametrize("scenario_id", list(SCENARIOS))
+def test_device_matches_reference_script(scenario_id, monkeypatch, tmp_path):
+    """Calibration and every move write the same PV sequence as the script."""
+    scenario = SCENARIOS[scenario_id]
+    expected = run_reference(scenario, monkeypatch, tmp_path)
+    actual = run_device(scenario)
+
+    for key, exp, act in zip(
+        ("laser_period", "offset_t", "offset_phase_t", "phase_t"),
+        expected["calibration"],
+        actual["calibration"],
+    ):
+        assert_same_value(exp, act, f"{scenario_id}: calibration {key}")
+
+    assert len(actual["moves"]) == len(expected["moves"])
+    for exp_move, act_move in zip(expected["moves"], actual["moves"]):
+        context = f"{scenario_id}: move to {exp_move['target']}"
+        assert_same_puts(exp_move["puts"], act_move["puts"], context)
+        # delay card first, then pairs of V1/V2; one script sleep per pair
+        pvs = [pv for pv, _ in exp_move["puts"]]
+        assert pvs[:2] == [DELAY, FINE]
+        assert pvs[2::2] == [V1] * exp_move["sleeps"] and pvs[3::2] == [V2] * exp_move["sleeps"]
+        for key, exp, act in zip(
+            ("laser_period", "offset_t", "offset_phase_t", "phase_t"),
+            exp_move["state"],
+            act_move["state"],
+        ):
+            assert_same_value(exp, act, f"{context}: stored {key}")
+        assert act_move["readback"] == exp_move["target"]
+
+
+@pytest.mark.parametrize("scenario_id", list(SCENARIOS))
+def test_shortest_phase_path_against_reference_script(scenario_id, monkeypatch, tmp_path):
+    """Opt-in phase_path="shortest": script delay card, folded ramps, documented wrap offset."""
+    scenario = SCENARIOS[scenario_id]
+    expected = run_reference(scenario, monkeypatch, tmp_path)
+    actual = run_device(scenario, phase_path="shortest")
+    for key, exp, act in zip(
+        ("laser_period", "offset_t", "offset_phase_t", "phase_t"),
+        expected["calibration"],
+        actual["calibration"],
+    ):
+        assert_same_value(exp, act, f"{scenario_id}: calibration {key}")
+
+    max_pairs = math.ceil(6.005 / (2.0 / 36)) + 1
+    touched = False
+    voltages = {"exp": (scenario.initial[V1], scenario.initial[V2])}
+    voltages["act"] = voltages["exp"]
+    stored = {"exp": expected["calibration"][3], "act": actual["calibration"][3]}
+    moves = iter(zip(expected["moves"], actual["moves"], strict=True))
+    for step in scenario.steps:
+        if isinstance(step, dict):  # both programs see the same hand-made change
+            voltages = {
+                key: (step.get(V1, v[0]), step.get(V2, v[1])) for key, v in voltages.items()
+            }
+            touched = True
+            continue
+        exp_move, act_move = next(moves)
+        context = f"{scenario_id}: move to {exp_move['target']}"
+        assert_same_puts(exp_move["puts"][:2], act_move["puts"][:2], f"{context}: delay card")
+        assert 1 <= (len(act_move["puts"]) - 2) // 2 <= max_pairs, context
+        for index in range(3):  # laser_period, offset_t, offset_phase_t
+            assert_same_value(exp_move["state"][index], act_move["state"][index], context)
+        assert act_move["readback"] == exp_move["target"]
+
+        wraps = (exp_move["state"][3] - act_move["state"][3]) / 12.01
+        assert wraps == pytest.approx(round(wraps), abs=1e-9), context
+        turned = act_move["state"][3] - stored["act"]
+        assert abs(turned) <= 6.005 + 1e-9, context
+        start = {key: phase_from_voltages(*v, 2.0) for key, v in voltages.items()}
+        voltages = {
+            "exp": (exp_move["puts"][-2][1], exp_move["puts"][-1][1]),
+            "act": (act_move["puts"][-2][1], act_move["puts"][-1][1]),
+        }
+        end = {key: phase_from_voltages(*v, 2.0) for key, v in voltages.items()}
+        assert wrap_phase(end["act"] - start["act"] - turned, 2.0) == pytest.approx(
+            0, abs=1e-9
+        ), context
+        if not touched:
+            assert wrap_phase(end["act"] - end["exp"], 2.0) == pytest.approx(
+                wrap_phase(-0.01 * round(wraps), 2.0), abs=1e-9
+            ), context
+        stored = {"exp": exp_move["state"][3], "act": act_move["state"][3]}
+
+
+def test_scenario_table_covers_the_edge_cases(monkeypatch, tmp_path):
+    """The scenarios really cover laser-period wraps, cavity crossings and long ramps."""
+    totals, phase_deltas, crossings = [], [], 0
+    for scenario in SCENARIOS.values():
+        expected = run_reference(scenario, monkeypatch, tmp_path)
+        laser_period, offset_t, _, phase_t = expected["calibration"]
+        targets = [move["target"] for move in expected["moves"]]
+        for previous, current in zip(targets, targets[1:]):
+            if math.floor(previous / 12.01) != math.floor(current / 12.01):
+                crossings += 1
+        for move in expected["moves"]:
+            totals.append((-offset_t - move["target"], laser_period))
+            phase_deltas.append(move["state"][3] - phase_t)
+            phase_t = move["state"][3]
+    assert any(total < 0 for total, _ in totals), "no case wraps below zero"
+    assert any(total >= period for total, period in totals), "no case wraps above the laser period"
+    assert crossings > 0, "no move crosses a cavity period"
+    assert min(phase_deltas) < -10 and max(phase_deltas) > 10, "no long ramps in both directions"
+
+
+def test_zero_phase_move_script_fails_device_writes_final_step(monkeypatch, tmp_path):
+    """Same delay twice: the script crashes in arange, the device writes only the final step."""
+    scenario = Scenario(STATE_A, 546.04 - 546.92, [0.5])
+    ioc = FakeIOC(scenario.initial)
+    script = load_reference_script(ioc, monkeypatch, tmp_path, [])
+    script.CurrentSituation(scenario.scope)
+    script.SetDelay(0.5)
+    ioc.puts.clear()
+    with pytest.raises((ValueError, ZeroDivisionError)):
+        script.SetDelay(0.5)
+    script_card_puts = list(ioc.puts)
+    assert [pv for pv, _ in script_card_puts] == [DELAY, FINE]
+
+    with patched_device(LaserDelay, name="laser_delay", step_sleep=0.0) as device:
+        set_pv_values(device, scenario.initial)
+        device.calibrate(scenario.scope)
+        device.set(0.5).wait(timeout=30)
+        log = record_puts(device)
+        v1_before, v2_before = device.phase_v1.get(), device.phase_v2.get()
+        status = device.set(0.5)
+        status.wait(timeout=5)
+        assert status.success
+        assert_same_puts(script_card_puts, log[:2], "repeat move: delay card")
+        assert [pv for pv, _ in log[2:]] == [V1, V2]
+        assert_same_value(v1_before, log[2][1], "repeat move: V1")
+        assert_same_value(v2_before, log[3][1], "repeat move: V2")
+
+
+def test_delay_card_conversion_matches_script_for_many_values(monkeypatch, tmp_path):
+    """int() truncations of MoveDelayCard, checked on many delays incl. near-integer boundaries."""
+    ioc = FakeIOC(STATE_A)
+    script = load_reference_script(ioc, monkeypatch, tmp_path, [])
+    rng = np.random.default_rng(20261013)
+    totals = list(rng.uniform(-5000.0, 5000.0, 3000))
+    # values whose fine part sits on a 0.01 ns boundary, where truncation is fragile
+    totals += [round(k * 0.01, 2) for k in range(0, 2001)]
+    totals += [k * 0.01 for k in range(-400, 400)]
+    for laser_period in (2000.2, 1000.0, 1000 * 2.0002, 12.01 * 83.0):
+        script.LaserPeriod = laser_period
+        for total in totals:
+            ioc.puts.clear()
+            script.MoveDelayCard(total)
+            (_, coarse), (_, fine) = ioc.puts  # pylint: disable=unbalanced-tuple-unpacking
+            assert delay_card_setting(total, laser_period) == (coarse, fine), (total, laser_period)
+
+
+def test_phase_ramp_matches_script_for_many_moves(monkeypatch, tmp_path):
+    """Ramp values (numpy.arange steps plus final step) of MovePhaseShifter for random moves."""
+    ioc = FakeIOC(STATE_A)
+    script = load_reference_script(ioc, monkeypatch, tmp_path, [])
+    rng = np.random.default_rng(42)
+    for _ in range(200):
+        angle = rng.uniform(-np.pi, np.pi)
+        ioc.values[V1], ioc.values[V2] = float(np.cos(angle)), float(np.sin(angle))
+        start = script.GetPhaseT()
+        relative = float(rng.uniform(-24.0, 24.0))
+        ioc.puts.clear()
+        script.MovePhaseShifter(relative)
+        expected = [value for _, value in ioc.puts]
+        actual = []
+        for offset in list(phase_ramp_offsets(relative, 2.0, 36)) + [relative]:
+            actual.extend(voltages_for_phase(start + offset, 2.0, 1.0))
+        assert len(actual) == len(expected)
+        for exp, act in zip(expected, actual):
+            assert_same_value(exp, act, f"phase ramp {relative}")
+
+
+def test_device_refuses_where_the_script_follows_touched_voltages(monkeypatch, tmp_path):
+    """Voltages turned by hand: the device refuses the move and writes nothing."""
+    scenario = Scenario(STATE_A, 546.04 - 546.92, [1.0, {V1: 0.0, V2: 1.0}, 2.0])
+    expected = run_reference(scenario, monkeypatch, tmp_path)
+    assert len(expected["moves"]) == 2  # the script moves twice
+    with patched_device(
+        LaserDelay, name="laser_delay", step_sleep=0.0, fine_rollover=False
+    ) as device:
+        set_pv_values(device, scenario.initial)
+        device.calibrate(scenario.scope)
+        device.set(1.0).wait(timeout=30)
+        set_pv_values(device, {V1: 0.0, V2: 1.0})
+        log = record_puts(device)
+        status = device.set(2.0)
+        with pytest.raises(LaserDelayHardwareError):
+            status.wait(timeout=30)
+        assert log == []

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)