fix(stereo-split): detect localized timestamp bursts

jobs/stereo-split/convert_mcap_stereo_h264.py

@@ -273,6 +273,8 @@ def video_message_times(self, sample: _TimestampSample) -> tuple[int, int]:
 _MIN_HEADER_GAP_NS = 1_000_000
 _MAX_HEADER_GAP_NS = 1_000_000_000
 _MIN_STALL_GAP_NS = 250_000_000
+_TIMESTAMP_REPAIR_WINDOW_GAPS = 300
+_MIN_BURST_GAP_RATIO = 0.2
 
 
 def _percentile(values: list[int], fraction: float) -> int:
@@ -281,6 +283,24 @@ def _percentile(values: list[int], fraction: float) -> int:
     return ordered[index]
 
 
+def _contains_bursty_window(
+    gaps: list[int], burst_limit: int, stall_limit: int
+) -> bool:
+    window_size = min(len(gaps), _TIMESTAMP_REPAIR_WINDOW_GAPS)
+    burst_flags = [gap < burst_limit for gap in gaps]
+    stall_flags = [gap >= stall_limit for gap in gaps]
+    burst_count = sum(burst_flags[:window_size])
+    stall_count = sum(stall_flags[:window_size])
+    for start in range(len(gaps) - window_size + 1):
+        if burst_count / window_size >= _MIN_BURST_GAP_RATIO and stall_count > 0:
+            return True
+        if start + window_size == len(gaps):
+            break
+        burst_count += burst_flags[start + window_size] - burst_flags[start]
+        stall_count += stall_flags[start + window_size] - stall_flags[start]
+    return False
+
+
 def _timestamp_axis_issue(
     times: list[int], header_times: list[int], median_header_gap: int
 ) -> str | None:
@@ -296,9 +316,8 @@ def _timestamp_axis_issue(
         return "rate_mismatch"
 
     burst_limit = median_header_gap // 2
-    burst_ratio = sum(gap < burst_limit for gap in gaps) / len(gaps)
     stall_limit = max(_MIN_STALL_GAP_NS, median_header_gap * 8)
-    if burst_ratio >= 0.2 and max(gaps) >= stall_limit:
+    if _contains_bursty_window(gaps, burst_limit, stall_limit):
         return "bursty"
     return None
 

jobs/stereo-split/tests/test_convert.py

@@ -349,6 +349,23 @@ def test_timestamp_plan_repairs_publish_only(self) -> None:
             [publish_times[0] + timestamp - header_times[0] for timestamp in header_times],
         )
 
+    def test_timestamp_plan_repairs_bursty_prefix_in_otherwise_healthy_file(self) -> None:
+        frame_count = 1_000
+        bursty_frames = 150
+        header_times = regular_frame_times(frame_count)
+        bursty_prefix = bursty_frame_times(bursty_frames)
+        log_times = bursty_prefix + [
+            bursty_prefix[-1] + header_time - header_times[bursty_frames - 1]
+            for header_time in header_times[bursty_frames:]
+        ]
+
+        plan = _build_timestamp_repair_plan(
+            timestamp_samples(header_times, log_times)
+        )
+
+        self.assertTrue(plan.applied)
+        self.assertEqual(plan.reason, "log_bursty,publish_bursty")
+
     def test_timestamp_plan_preserves_healthy_outer_times(self) -> None:
         header_times = regular_frame_times()
         log_times = [timestamp + 5_000_000 for timestamp in header_times]