fix(timepix): publish NaN for on/off windows outside the received time bins

superxas_bec/devices/timepix/timepix.py

@@ -167,6 +167,27 @@ def load_pixel_map_from_json(file_path: str) -> PixelMap:
     return pixel_map
 
 
+def window_fits(window: slice, length: int) -> bool:
+    """True if [start, stop) is a non-empty index window inside range(length)."""
+    return 0 <= window.start < window.stop <= length
+
+
+def window_mean(data: np.ndarray, window: slice) -> float:
+    """
+    Mean of data over the half-open index window [start, stop).
+
+    Returns NaN if the window is empty or does not fit into data. numpy would instead average
+    over fewer points (stop past the end) or return NaN with a RuntimeWarning (empty window).
+
+    Args:
+        data (np.ndarray): 1D data, e.g. one time spectrum.
+        window (slice): window with integer start and stop.
+    """
+    if not window_fits(window, len(data)):
+        return float("nan")
+    return float(np.mean(data[window]))
+
+
 class ImagePlugin_Timepix(ImagePlugin_V35):
     """Custom Image Plugin for TimePix detector."""
 
@@ -306,6 +327,8 @@ class Timepix(PSIDeviceBase, TimePixControl):
         # Settings that timepix_scan overrides in on_stage, saved for the restore in
         # on_unstage / on_stop. None while no timepix_scan overrides are active.
         self._settings_before_timepix_scan: dict[str, Any] | None = None
+        # Log the "on/off window does not fit" warning only once per scan.
+        self._on_off_window_warned = False
 
     def stage(self) -> list[object] | StatusBase:  # type: ignore
         """Stage the device.
@@ -456,27 +479,55 @@ class Timepix(PSIDeviceBase, TimePixControl):
             # Frames can arrive without a staged timepix_scan (e.g. before the first stage after
             # a restart), so scan_parameters may still be None here.
             if self._is_timepix_scan():
-                params = self.scan_parameters.additional_scan_parameters
-                idx_off_start = params["idx_off_start"]
-                idx_off_stop = params["idx_off_stop"]
-                idx_on_start = params["idx_on_start"]
-                idx_on_stop = params["idx_on_stop"]
-                level_1_off = np.mean(data_1[idx_off_start:idx_off_stop])
-                level_1_on = np.mean(data_1[idx_on_start:idx_on_stop])
-                level_2_off = np.mean(data_2[idx_off_start:idx_off_stop])
-                level_2_on = np.mean(data_2[idx_on_start:idx_on_stop])
-                self.xes_level_1_off.put(level_1_off)
-                self.xes_level_1_on.put(level_1_on)
-                self.xes_level_2_off.put(level_2_off)
-                self.xes_level_2_on.put(level_2_on)
-                self.xes_diff_1_on_off.put(level_1_on - level_1_off)
-                self.xes_diff_2_on_off.put(level_2_on - level_2_off)
+                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 _put_on_off_levels(self, data_1: np.ndarray, data_2: np.ndarray) -> None:
+        """
+        Publish the laser on/off levels of a timepix_scan for both energy groups.
+
+        The off/on windows are the half-open time-bin ranges [idx_*_start, idx_*_stop) given as
+        timepix_scan arguments. A level is the mean of xes_energy_1/2 over its window, the
+        difference is on minus off. If a window is empty or does not fit into the received
+        time bins (TimePixFly sent fewer bins than time_roi_n), its levels and the differences
+        are published as NaN, so every frame still adds one value per signal and the signals stay
+        aligned with the scan points; a warning is logged once per scan.
+
+        Args:
+            data_1 (np.ndarray): xes_energy_1 of this frame, one value per time bin.
+            data_2 (np.ndarray): xes_energy_2 of this frame, one value per time bin.
+        """
+        params = self.scan_parameters.additional_scan_parameters
+        off = slice(params["idx_off_start"], params["idx_off_stop"])
+        on = slice(params["idx_on_start"], params["idx_on_stop"])
+
+        bad_windows = [
+            f"{name} [{window.start}, {window.stop})"
+            for name, window in (("off", off), ("on", on))
+            if not window_fits(window, len(data_1))
+        ]
+        if bad_windows and not self._on_off_window_warned:
+            self._on_off_window_warned = True
+            logger.warning(
+                f"{self.name}: laser {' and '.join(bad_windows)} window does not fit into the "
+                f"{len(data_1)} time bins received; publishing NaN for the affected levels."
+            )
+
+        level_1_off = window_mean(data_1, off)
+        level_1_on = window_mean(data_1, on)
+        level_2_off = window_mean(data_2, off)
+        level_2_on = window_mean(data_2, on)
+        self.xes_level_1_off.put(level_1_off)
+        self.xes_level_1_on.put(level_1_on)
+        self.xes_level_2_off.put(level_2_off)
+        self.xes_level_2_on.put(level_2_on)
+        self.xes_diff_1_on_off.put(level_1_on - level_1_off)
+        self.xes_diff_2_on_off.put(level_2_on - level_2_off)
+
     def _is_timepix_scan(self) -> bool:
         """True if the device is staged for a timepix_scan (False before the first stage)."""
         return (
@@ -683,6 +734,7 @@ class Timepix(PSIDeviceBase, TimePixControl):
 
         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:

tests/tests_devices/test_timepix.py

@@ -26,6 +26,7 @@ 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 (
@@ -524,6 +525,43 @@ def test_timepix_on_stop_stops_camera_writer_and_backend(timepix):
     timepix.backend.on_stop.assert_called_once()
 
 
+@pytest.mark.parametrize(
+    "window, expected",
+    [
+        (slice(0, 2), 1.5),
+        (slice(1, 4), 3.0),
+        (slice(2, 2), np.nan),  # empty
+        (slice(3, 2), np.nan),  # start after stop
+        (slice(2, 5), np.nan),  # stop past the end
+        (slice(4, 5), np.nan),  # start past the end
+        (slice(-1, 2), np.nan),  # negative start
+    ],
+)
+def test_window_mean(window, expected):
+    data = np.array([1.0, 2.0, 3.0, 4.0], dtype=np.float32)
+    np.testing.assert_equal(window_mean(data, window), expected)
+
+
+def test_timepix_on_off_levels_are_nan_for_window_outside_received_bins(timepix):
+    """If TimePixFly sends fewer time bins than requested, the affected levels become NaN."""
+    _set_scan_parameters(timepix, "timepix_scan", TIMEPIX_SCAN_PARAMETERS)
+    timepix.troin = 4
+    # only 3 time bins: xes_energy_1 = [4, 8, 12], xes_energy_2 = [40, 80, 120]
+    frames = _xes_frames(_level_test_spectra()[:3])
+
+    with mock.patch("superxas_bec.devices.timepix.timepix.logger") as logger:
+        timepix.msg_buffer_callback(*frames)
+        timepix.msg_buffer_callback(*frames)
+
+    assert _message_value(timepix.xes_level_1_off) == pytest.approx(6.0)
+    assert _message_value(timepix.xes_level_2_off) == pytest.approx(60.0)
+    for name in ("xes_level_1_on", "xes_level_2_on", "xes_diff_1_on_off", "xes_diff_2_on_off"):
+        assert np.isnan(_message_value(getattr(timepix, name))), name
+    # one warning per scan, not one per frame
+    assert logger.warning.call_count == 1
+    assert "on [2, 4)" in logger.warning.call_args.args[0]
+
+
 def _stage_for(dev, scan_name: str, additional_scan_parameters: dict | None = None, **kwargs):
     """Stage the device for a scan, as the device server does (stage(), not on_stage())."""
     dev.scan_info.msg = _scan_status_msg(scan_name, additional_scan_parameters, **kwargs)