Fix WindowAggregatingExtractor to handle timing jitter robustly

With timing noise (e.g., frames at [0.0001, 1, 2, 3, 4, 5]), a 5-second
window would incorrectly include 6 frames instead of 5 due to the
inclusive lower bound in label-based slicing.

Solution:
- Estimate frame period from median interval between frames
- Shift cutoff by +0.5 × median_interval to place window boundary
  between frame slots, avoiding extra frames from timing jitter
- Clamp cutoff to latest_time for narrow windows (duration < median_interval)
- Continue using inclusive label-based slicing: data[time, cutoff:]

This automatically adapts to different frame rates and handles both
timing jitter and narrow windows correctly.

Add comprehensive tests for timing jitter scenarios:
- test_handles_timing_jitter_at_window_start
- test_handles_timing_jitter_at_window_end
- test_consistent_frame_count_with_perfect_timing

Original prompt: Please think about a conceptualy problem in WindowAggregatingExtractor:
- Data arrives in regular (but of source noisy) intervals, say once per second.
- User requests 5 second sliding window.
- Current extraction code will then often return to many frames.
Example:
Frames at [0.0001, 1,2,3,4,5]
=> we get 6 frames instead of 5. (I think even with 0.0 it is wrong, since label-based indexing used to extracted windowed_data is inclusive on the left).

The problem is that we can't simply reduce to, say, 4 seconds, or 5 frames, since frame rates can vary a lot. Can you think of a more stable approach?

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <[email protected]>

Author: SimonHeybrock

src/ess/livedata/dashboard/extractors.py

@@ -129,7 +129,24 @@ def extract(self, data: sc.DataArray) -> Any:
         duration = sc.scalar(self._window_duration_seconds, unit='s').to(
             unit=time_coord.unit
         )
-        windowed_data = data[self._concat_dim, latest_time - duration :]
+
+        # Estimate frame period from median interval to handle timing noise.
+        # Shift cutoff by half period to place boundary between frame slots,
+        # avoiding inclusion of extra frames due to timing jitter.
+        if len(time_coord) > 1:
+            intervals = time_coord[1:] - time_coord[:-1]
+            median_interval = sc.median(intervals)
+            cutoff_time = latest_time - duration + 0.5 * median_interval
+            # Clamp to ensure at least latest frame included
+            # (handles narrow windows where duration < median_interval)
+            if cutoff_time > latest_time:
+                cutoff_time = latest_time
+        else:
+            # Single frame: use duration-based cutoff
+            cutoff_time = latest_time - duration
+
+        # Use label-based slicing with inclusive lower bound
+        windowed_data = data[self._concat_dim, cutoff_time:]
 
         # Resolve and cache aggregator function on first call
         if self._aggregator is None:

tests/dashboard/extractors_test.py

@@ -413,6 +413,97 @@ def test_extract_narrow_window(self):
             result.data, sc.array(dims=['x'], values=[5, 6], unit='counts')
         )
 
+    def test_handles_timing_jitter_at_window_start(self):
+        """Test that timing noise near window boundary doesn't include extra frames."""
+        extractor = WindowAggregatingExtractor(
+            window_duration_seconds=5.0,
+            aggregation=WindowAggregation.nansum,
+            concat_dim='time',
+        )
+
+        # Regular 1 Hz data with timing jitter on first frame
+        # Conceptually frames at t=[0, 1, 2, 3, 4, 5] but first has noise
+        data = sc.DataArray(
+            sc.array(
+                dims=['time', 'x'],
+                values=[[1, 2], [3, 4], [5, 6], [7, 8], [9, 10], [11, 12]],
+                unit='counts',
+            ),
+            coords={
+                'time': sc.array(
+                    dims=['time'], values=[0.0001, 1.0, 2.0, 3.0, 4.0, 5.0], unit='s'
+                ),
+                'x': sc.arange('x', 2, unit='m'),
+            },
+        )
+
+        result = extractor.extract(data)
+
+        # Window (5-5, 5] = (0, 5] excludes frame at 0.0001 (using exclusive bound)
+        # Should include 5 frames [1, 2, 3, 4, 5], not all 6
+        expected_sum = sc.array(dims=['x'], values=[35, 40], unit='counts')
+        assert sc.allclose(result.data, expected_sum)
+
+    def test_handles_timing_jitter_at_window_end(self):
+        """Test that timing noise on latest frame doesn't affect frame count."""
+        extractor = WindowAggregatingExtractor(
+            window_duration_seconds=5.0,
+            aggregation=WindowAggregation.nansum,
+            concat_dim='time',
+        )
+
+        # Regular 1 Hz data with timing jitter on last frame
+        data = sc.DataArray(
+            sc.array(
+                dims=['time', 'x'],
+                values=[[1, 2], [3, 4], [5, 6], [7, 8], [9, 10], [11, 12]],
+                unit='counts',
+            ),
+            coords={
+                'time': sc.array(
+                    dims=['time'], values=[0.0, 1.0, 2.0, 3.0, 4.0, 5.0001], unit='s'
+                ),
+                'x': sc.arange('x', 2, unit='m'),
+            },
+        )
+
+        result = extractor.extract(data)
+
+        # Window (5.0001-5, 5.0001] = (0.0001, 5.0001]
+        # Should include 5 frames [1, 2, 3, 4, 5.0001]
+        expected_sum = sc.array(dims=['x'], values=[35, 40], unit='counts')
+        assert sc.allclose(result.data, expected_sum)
+
+    def test_consistent_frame_count_with_perfect_timing(self):
+        """Test baseline: perfect timing gives expected frame count."""
+        extractor = WindowAggregatingExtractor(
+            window_duration_seconds=5.0,
+            aggregation=WindowAggregation.nansum,
+            concat_dim='time',
+        )
+
+        # Perfect 1 Hz data at exactly [0, 1, 2, 3, 4, 5]
+        data = sc.DataArray(
+            sc.array(
+                dims=['time', 'x'],
+                values=[[1, 2], [3, 4], [5, 6], [7, 8], [9, 10], [11, 12]],
+                unit='counts',
+            ),
+            coords={
+                'time': sc.array(
+                    dims=['time'], values=[0.0, 1.0, 2.0, 3.0, 4.0, 5.0], unit='s'
+                ),
+                'x': sc.arange('x', 2, unit='m'),
+            },
+        )
+
+        result = extractor.extract(data)
+
+        # Window (0, 5] excludes frame at exactly 0 (exclusive bound)
+        # Should include 5 frames [1, 2, 3, 4, 5]
+        expected_sum = sc.array(dims=['x'], values=[35, 40], unit='counts')
+        assert sc.allclose(result.data, expected_sum)
+
 
 class TestCreateExtractorsFromParams:
     """Tests for create_extractors_from_params factory function."""