Fix dejittering of timestamps

CHANGELOG.md

@@ -1,4 +1,6 @@
 ## [UNRELEASED] - YYYY-MM-DD
+### Fixed
+- Skip linear dejittering for streams flagged with `can_drop_samples` to prevent dropped-frame streams from being shifted too early in time ([#165](https://github.com/xdf-modules/xdf-Python/pull/165) by [Clemens Brunner](https://github.com/cbrnr))
 
 ## [1.17.3] - 2026-01-20
 ### Fixed

src/pyxdf/pyxdf.py

@@ -380,6 +380,7 @@ def load_xdf(
             temp,
             jitter_break_threshold_seconds,
             jitter_break_threshold_samples,
+            stream_headers=streams,
         )
     else:
         for stream in temp.values():
@@ -899,7 +900,37 @@ def _detect_breaks(stream, threshold_seconds=1.0, threshold_samples=500):
     return b_breaks
 
 
-def _jitter_removal(streams, threshold_seconds=1, threshold_samples=500):
+def _stream_can_drop_samples(stream_meta):
+    """Return True if stream metadata indicates samples can be dropped."""
+    if not stream_meta:
+        return False
+
+    info = stream_meta.get("info", {})
+    desc = info.get("desc")
+    if not isinstance(desc, list) or len(desc) == 0 or not isinstance(desc[0], dict):
+        return False
+
+    synchronization = desc[0].get("synchronization")
+    if (
+        not isinstance(synchronization, list)
+        or len(synchronization) == 0
+        or not isinstance(synchronization[0], dict)
+    ):
+        return False
+
+    can_drop_samples = synchronization[0].get("can_drop_samples")
+    if not isinstance(can_drop_samples, list) or len(can_drop_samples) == 0:
+        return False
+
+    return str(can_drop_samples[0]).lower() == "true"
+
+
+def _jitter_removal(
+    streams,
+    threshold_seconds=1,
+    threshold_samples=500,
+    stream_headers=None,
+):
     for stream_id, stream in streams.items():
         stream.effective_srate = 0  # will be recalculated if possible
         nsamples = len(stream.time_stamps)
@@ -909,6 +940,19 @@ def _jitter_removal(streams, threshold_seconds=1, threshold_samples=500):
                 stream.segments.append((0, nsamples - 1))  # inclusive
                 continue
 
+            # Streams that can drop samples (e.g., video/HMD) should not undergo linear
+            # dejittering because it compresses dropped-frame intervals and can shift
+            # the segment start substantially earlier.
+            if _stream_can_drop_samples(
+                None if stream_headers is None else stream_headers.get(stream_id)
+            ):
+                stream.segments.append((0, nsamples - 1))  # inclusive
+                if nsamples > 1:
+                    duration = stream.time_stamps[-1] - stream.time_stamps[0]
+                    if duration > 0:
+                        stream.effective_srate = (nsamples - 1) / duration
+                continue
+
             # Find boundary breaks
             b_breaks = _detect_breaks(stream, threshold_seconds, threshold_samples)
             # Find segment indices

test/test_jitter_removal.py

@@ -1,9 +1,9 @@
 import numpy as np
 import pytest
-from pyxdf.pyxdf import _jitter_removal
-
 from mock_data_stream import MockStreamData
 
+from pyxdf.pyxdf import _jitter_removal
+
 
 @pytest.mark.parametrize(
     "t_start, t_end",
@@ -133,3 +133,47 @@ def test_jitter_removal_with_jitter(t_start, t_end):
     )
     np.testing.assert_equal(stream.time_series[:, 0], time_stamps)
     np.testing.assert_allclose(stream.effective_srate, srate)
+
+
+def test_jitter_removal_can_drop_samples_preserves_timing():
+    srate = 90
+    tdiff = 1 / srate
+
+    # Simulate regular frame drops in a nominally regular stream. Applying linear
+    # dejittering to this data would wrongly compress dropped-frame intervals and shift
+    # the start too early.
+    n_samples_nominal = 1800
+    nominal = np.arange(n_samples_nominal) * tdiff
+    keep_mask = np.ones(n_samples_nominal, dtype=bool)
+    keep_mask[::5] = False
+    time_stamps = nominal[keep_mask]
+
+    streams = {1: MockStreamData(time_stamps=time_stamps, srate=srate, tdiff=tdiff)}
+    stream_headers = {
+        1: {
+            "info": {
+                "desc": [
+                    {
+                        "synchronization": [
+                            {
+                                "can_drop_samples": ["true"],
+                            }
+                        ]
+                    }
+                ]
+            }
+        }
+    }
+
+    _jitter_removal(
+        streams,
+        threshold_seconds=1,
+        threshold_samples=500,
+        stream_headers=stream_headers,
+    )
+
+    stream = streams[1]
+    assert stream.segments == [(0, len(time_stamps) - 1)]
+    np.testing.assert_allclose(stream.time_stamps, time_stamps, atol=1e-14)
+    np.testing.assert_equal(stream.time_series[:, 0], time_stamps)
+    assert stream.effective_srate < srate