fix: keep subtitle timing continuous across concatenated audio clips

funclip/videoclipper.py

@@ -221,23 +221,32 @@ def clip(self, dest_text, start_ost, end_ost, state, dest_spk=None, output_dir=N
         ts = all_ts
         # ts.sort()
         srt_index = 0
+        # Output position in samples; kept as an integer and converted per call so
+        # summing float durations cannot drift a cue below its millisecond.
+        acc_samples = 0
         clip_srt = ""
         if len(ts):
             start, end = ts[0]
             start = min(max(0, start+start_ost*16), len(data))
             end = min(max(0, end+end_ost*16), len(data))
             res_audio = data[start:end]
             start_end_info = "from {} to {}".format(start/16000, end/16000)
-            srt_clip, _, srt_index = generate_srt_clip(sentences, start/16000.0, end/16000.0, begin_index=srt_index)
+            srt_clip, _, srt_index = generate_srt_clip(sentences, start/16000.0, end/16000.0, begin_index=srt_index, time_acc_ost=acc_samples / 16000.0)
             clip_srt += srt_clip
+            # Each later region is appended after the audio already concatenated,
+            # so its subtitles start at that output time, not at zero. Count the
+            # samples actually appended: offsets can clamp a region to start > end,
+            # which appends nothing.
+            acc_samples += len(data[start:end])
             for _ts in ts[1:]:  # multiple sentence input or multiple output matched
                 start, end = _ts
                 start = min(max(0, start+start_ost*16), len(data))
                 end = min(max(0, end+end_ost*16), len(data))
                 start_end_info += ", from {} to {}".format(start, end)
                 res_audio = np.concatenate([res_audio, data[start:end]], -1)
-                srt_clip, _, srt_index = generate_srt_clip(sentences, start/16000.0, end/16000.0, begin_index=srt_index-1)
+                srt_clip, _, srt_index = generate_srt_clip(sentences, start/16000.0, end/16000.0, begin_index=srt_index-1, time_acc_ost=acc_samples / 16000.0)
                 clip_srt += srt_clip
+                acc_samples += len(data[start:end])
         if len(ts):
             message = "{} periods found in the speech: ".format(len(ts)) + start_end_info + log_append
         else:

tests/test_audio_clip_subtitle_timeline.py

@@ -0,0 +1,146 @@
+import copy
+import re
+import sys
+import unittest
+from pathlib import Path
+
+import numpy as np
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "funclip"))
+
+from videoclipper import VideoClipper  # noqa: E402
+
+
+CUE_TIMES = re.compile(r"(\d\d:\d\d:\d\d,\d\d\d) --> (\d\d:\d\d:\d\d,\d\d\d)")
+
+
+class TestAudioClipSubtitleTimeline(unittest.TestCase):
+    """Subtitles of concatenated audio regions follow the concatenated output."""
+
+    def setUp(self):
+        sentences = [
+            {"text": "hello", "timestamp": [[1000, 2000]], "spk": 0},
+            {"text": "world", "timestamp": [[4000, 5000]], "spk": 0},
+        ]
+        self.state = {
+            "audio_input": (16000, np.zeros(96000)),
+            "recog_res_raw": "hello world",
+            "timestamp": [[1000, 2000], [4000, 5000]],
+            "sentences": sentences,
+            "sd_sentences": copy.deepcopy(sentences),
+        }
+
+    def clip(self, *args, **kwargs):
+        return VideoClipper(None).clip(*args, state=copy.deepcopy(self.state), **kwargs)
+
+    def test_second_region_subtitle_starts_after_the_first_region(self):
+        (rate, audio), _, subtitles = self.clip("hello#world", 0, 0)
+
+        self.assertEqual((rate, len(audio)), (16000, 32000))
+        self.assertEqual(
+            CUE_TIMES.findall(subtitles),
+            [("00:00:00,000", "00:00:01,000"), ("00:00:01,000", "00:00:02,000")],
+        )
+
+    def test_explicit_timestamps_follow_the_concatenated_output(self):
+        (_, audio), _, subtitles = self.clip(
+            None, 0, 0, timestamp_list=[[16000, 32000], [64000, 80000]]
+        )
+
+        self.assertEqual(len(audio), 32000)
+        self.assertEqual(
+            CUE_TIMES.findall(subtitles),
+            [("00:00:00,000", "00:00:01,000"), ("00:00:01,000", "00:00:02,000")],
+        )
+
+    def test_single_region_is_unchanged(self):
+        (_, audio), _, subtitles = self.clip("hello", 0, 0)
+
+        self.assertEqual(len(audio), 16000)
+        self.assertEqual(CUE_TIMES.findall(subtitles), [("00:00:00,000", "00:00:01,000")])
+
+    def clip_with(self, sentences, seconds, *args, **kwargs):
+        state = {
+            "audio_input": (16000, np.zeros(16000 * seconds)),
+            "recog_res_raw": "",
+            "timestamp": [],
+            "sentences": sentences,
+            "sd_sentences": copy.deepcopy(sentences),
+        }
+        return VideoClipper(None).clip(*args, state=state, **kwargs)
+
+    def test_region_emptied_by_start_offset_does_not_shift_later_subtitles(self):
+        # start_ost=1500ms turns the first region into 2.5-2.0s (nothing appended);
+        # the second contributes 5.5-7.0s, so "world" (6-7s) sits at 0.5-1.5s.
+        sentences = [
+            {"text": "hello", "timestamp": [[1000, 2000]], "spk": 0},
+            {"text": "world", "timestamp": [[6000, 7000]], "spk": 0},
+        ]
+        (_, audio), _, subtitles = self.clip_with(
+            sentences, 8, None, 1500, 0, timestamp_list=[[16000, 32000], [64000, 112000]]
+        )
+
+        self.assertEqual(len(audio), 24000)
+        self.assertEqual(CUE_TIMES.findall(subtitles), [("00:00:00,500", "00:00:01,500")])
+
+    def test_middle_region_emptied_by_start_offset_adds_no_time(self):
+        # start_ost=1000ms: the first region becomes 1-3s, the middle one 5-4.5s
+        # (start > end, nothing appended) and the last 7-8s.
+        sentences = [
+            {"text": "one", "timestamp": [[2000, 2800]], "spk": 0},
+            {"text": "two", "timestamp": [[7200, 7800]], "spk": 0},
+        ]
+        (_, audio), _, subtitles = self.clip_with(
+            sentences,
+            8,
+            None,
+            1000,
+            0,
+            timestamp_list=[[0, 48000], [64000, 72000], [96000, 128000]],
+        )
+
+        self.assertEqual(len(audio), 48000)
+        self.assertEqual(
+            CUE_TIMES.findall(subtitles),
+            [("00:00:01,000", "00:00:01,800"), ("00:00:02,200", "00:00:02,800")],
+        )
+
+    def test_end_offset_clamped_to_audio_length(self):
+        # end_ost=2000ms extends the first region to 0-3s and would push the
+        # second past the 4s input; it is clamped to 3-4s.
+        sentences = [
+            {"text": "one", "timestamp": [[0, 1000]], "spk": 0},
+            {"text": "two", "timestamp": [[3000, 4000]], "spk": 0},
+        ]
+        (_, audio), _, subtitles = self.clip_with(
+            sentences, 4, None, 0, 2000, timestamp_list=[[0, 16000], [48000, 64000]]
+        )
+
+        self.assertEqual(len(audio), 64000)
+        self.assertEqual(CUE_TIMES.findall(subtitles)[-1], ("00:00:03,000", "00:00:04,000"))
+
+    def test_accumulated_offset_does_not_lose_a_millisecond_to_float_sums(self):
+        # 114240 + 105600 samples is exactly 13.74s, but 7.14 + 6.6 in floats is
+        # 13.739999..., which the millisecond formatter truncated to 13.739s.
+        seconds = 20
+        sentences = [
+            {"text": "a", "timestamp": [[0, 7140]], "spk": 0},
+            {"text": "b", "timestamp": [[8000, 14600]], "spk": 0},
+            {"text": "c", "timestamp": [[15000, 16000]], "spk": 0},
+        ]
+        (_, audio), _, subtitles = self.clip_with(
+            sentences,
+            seconds,
+            None,
+            0,
+            0,
+            timestamp_list=[[0, 114240], [128000, 233600], [240000, 256000]],
+        )
+
+        self.assertEqual(len(audio), 114240 + 105600 + 16000)
+        self.assertEqual(CUE_TIMES.findall(subtitles)[2], ("00:00:13,740", "00:00:14,740"))
+
+
+if __name__ == "__main__":
+    unittest.main()