From ddef0448bf3300d4d27eb5d94b5b5e7eb0213091 Mon Sep 17 00:00:00 2001 From: wyzula-jan Date: Sat, 3 Oct 2026 16:40:55 +0200 Subject: [PATCH] fix(timepix): publish NaN for on/off windows outside the received time bins --- superxas_bec/devices/timepix/timepix.py | 82 ++++++++++++++++++++----- tests/tests_devices/test_timepix.py | 38 ++++++++++++ 2 files changed, 105 insertions(+), 15 deletions(-) diff --git a/superxas_bec/devices/timepix/timepix.py b/superxas_bec/devices/timepix/timepix.py index fdea1d6..3481515 100644 --- a/superxas_bec/devices/timepix/timepix.py +++ b/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: diff --git a/tests/tests_devices/test_timepix.py b/tests/tests_devices/test_timepix.py index f7c943a..86576a3 100644 --- a/tests/tests_devices/test_timepix.py +++ b/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)