fix: skip forced and sparse embedded subtitle streams in subs_then_* VAD modes

HISTORY.rst

@@ -1,6 +1,10 @@
 History
 =======
 
+Unreleased
+----------
+* In the ``subs_then_*`` VAD modes, skip embedded subtitle streams flagged ``forced``, pick the reference stream by cue count instead of last cue time, and fall back to audio when the best stream has fewer than ``--min-embedded-subtitle-cues`` cues (default 50). Forced-only stubs previously produced offsets pinned at ``--max-offset-seconds`` with negative scores on in-sync subtitles (#238);
+
 0.5.1 (2026-07-23)
 ------------------
 * Make faust-cchardet a required dependency now that it works on Python 3.13+;

ffsubsync/constants.py

@@ -16,6 +16,11 @@
 DEFAULT_SCALE_FACTOR: float = 1
 DEFAULT_VAD: str = "subs_then_webrtc"
 DEFAULT_MAX_OFFSET_SECONDS: int = 60
+# In the subs_then_* VAD modes, an embedded subtitle stream with fewer cues than
+# this is treated as absent so alignment falls back to audio. Forced-only tracks
+# (signs and foreign lines) are typically 1 to 60 cues and carry too little
+# speech to align against.
+DEFAULT_MIN_EMBEDDED_SUBTITLE_CUES: int = 50
 DEFAULT_APPLY_OFFSET_SECONDS: int = 0
 
 # Penalty applied by the alass-style piecewise aligner (--split-penalty) when the

ffsubsync/ffsubsync.py

@@ -16,6 +16,7 @@
     DEFAULT_APPLY_OFFSET_SECONDS,
     DEFAULT_FRAME_RATE,
     DEFAULT_MAX_OFFSET_SECONDS,
+    DEFAULT_MIN_EMBEDDED_SUBTITLE_CUES,
     DEFAULT_MAX_SUBTITLE_SECONDS,
     DEFAULT_NON_SPEECH_LABEL,
     DEFAULT_SPLIT_PENALTY,
@@ -482,6 +483,11 @@ def make_reference_pipe(
                         extract_audio_first=getattr(
                             args, "extract_audio_first", False
                         ),
+                        min_embedded_subtitle_cues=getattr(
+                            args,
+                            "min_embedded_subtitle_cues",
+                            DEFAULT_MIN_EMBEDDED_SUBTITLE_CUES,
+                        ),
                         progress_handler=progress_handler,
                     ),
                 ),
@@ -1103,6 +1109,16 @@ def add_cli_only_args(parser: argparse.ArgumentParser) -> None:
             default=DEFAULT_VAD, choices=", ".join(VAD_CHOICES)
         ),
     )
+    parser.add_argument(
+        "--min-embedded-subtitle-cues",
+        type=int,
+        default=DEFAULT_MIN_EMBEDDED_SUBTITLE_CUES,
+        help="With a subs_then_* VAD, ignore embedded subtitle streams that have "
+        "fewer cues than this and align against audio instead (default=%d). "
+        "Forced-only tracks (signs, foreign lines) are usually far below this "
+        "and produce spurious offsets when used as the reference. Set to 0 to "
+        "restore the old behavior." % DEFAULT_MIN_EMBEDDED_SUBTITLE_CUES,
+    )
     parser.add_argument(
         "--whisper-weights",
         "--whisper-model",

ffsubsync/speech_transformers.py

@@ -18,6 +18,7 @@
 from ffsubsync.constants import (
     DEFAULT_ENCODING,
     DEFAULT_MAX_SUBTITLE_SECONDS,
+    DEFAULT_MIN_EMBEDDED_SUBTITLE_CUES,
     DEFAULT_SCALE_FACTOR,
     DEFAULT_START_SECONDS,
     DEFAULT_WHISPER_LANGUAGE,
@@ -264,9 +265,13 @@ def _probe_embedded_subtitle_streams(
     so that every subtitle stream can be extracted in a single ffmpeg pass.
     Bitmap subtitle codecs (PGS, VobSub, DVB, ...) are skipped because they
     cannot be muxed to SRT and would otherwise abort the whole extraction.
+    Streams flagged ``forced`` are skipped too: they hold only signs and
+    foreign-language lines, far too little speech to align against.
 
     Returns ``None`` when ffprobe is unavailable or fails, signaling the
-    caller to fall back to extracting streams one at a time.
+    caller to fall back to extracting streams one at a time. Returns an empty
+    list when ffprobe worked but found no usable stream, so the caller does
+    not go on to probe (and use) the streams that were deliberately skipped.
     """
     ffprobe_args = [
         ffmpeg_bin_path("ffprobe", gui_mode, ffmpeg_resources_path=ffmpeg_path),
@@ -275,7 +280,7 @@ def _probe_embedded_subtitle_streams(
         "-select_streams",
         "s",
         "-show_entries",
-        "stream=index,codec_name",
+        "stream=index,codec_name:stream_disposition=forced",
         "-of",
         "csv=p=0",
         fname,
@@ -295,10 +300,14 @@ def _probe_embedded_subtitle_streams(
             continue
         index = parts[0]
         codec_name = parts[1].strip().lower() if len(parts) > 1 else ""
+        forced = len(parts) > 2 and parts[2].strip() == "1"
         if codec_name in _BITMAP_SUBTITLE_CODECS:
             continue
+        if forced:
+            logger.info("skipping forced embedded subtitle stream 0:%s", index)
+            continue
         streams.append("0:{}".format(index))
-    return streams or None
+    return streams
 
 
 _FUSION_STRATEGIES: Tuple[str, ...] = ("weighted", "intersection", "union")
@@ -383,9 +392,11 @@ def __init__(
         max_duration_seconds: Optional[float] = None,
         extract_audio_first: bool = False,
         progress_handler: Optional[Callable[["ProgressInfo"], None]] = None,
+        min_embedded_subtitle_cues: int = DEFAULT_MIN_EMBEDDED_SUBTITLE_CUES,
     ) -> None:
         super(VideoSpeechTransformer, self).__init__()
         self.vad: str = vad
+        self.min_embedded_subtitle_cues: int = min_embedded_subtitle_cues
         self.sample_rate: int = sample_rate
         self.frame_rate: int = frame_rate
         self._non_speech_label: float = non_speech_label
@@ -408,15 +419,15 @@ def _probe_embedded_subtitle_streams(self, fname: str) -> Optional[List[str]]:
 
     def _extract_embedded_subs_single_pass(
         self, fname: str, streams: List[str]
-    ) -> Optional[List[io.BytesIO]]:
+    ) -> Optional[List[Tuple[str, io.BytesIO]]]:
         """Extract several subtitle streams in one ffmpeg invocation.
 
         ffmpeg can only send a single output to stdout, so each stream is
         written to a temporary file (in the system temp dir -- never next to
         the source media), read back into memory, and then deleted along with
-        the temp dir. Returns one in-memory buffer per stream that produced
-        data, or ``None`` if the ffmpeg invocation failed wholesale (the caller
-        then falls back to per-stream extraction).
+        the temp dir. Returns one ``(stream, buffer)`` pair per stream that
+        produced data, or ``None`` if the ffmpeg invocation failed wholesale
+        (the caller then falls back to per-stream extraction).
         """
         with tempfile.TemporaryDirectory(prefix="ffsubsync_subs_") as tmpdir:
             ffmpeg_args = [
@@ -442,24 +453,25 @@ def _extract_embedded_subs_single_pass(
             process.communicate()
             if process.returncode != 0:
                 return None
-            buffers: List[io.BytesIO] = []
-            for out_path in out_paths:
+            buffers: List[Tuple[str, io.BytesIO]] = []
+            for stream, out_path in zip(streams, out_paths):
                 if not os.path.exists(out_path) or os.path.getsize(out_path) == 0:
                     continue
                 with open(out_path, "rb") as f:
-                    buffers.append(io.BytesIO(f.read()))
+                    buffers.append((stream, io.BytesIO(f.read())))
             return buffers
 
     def _extract_embedded_subs_per_stream(
         self, fname: str, streams: List[str]
-    ) -> List[io.BytesIO]:
+    ) -> List[Tuple[str, io.BytesIO]]:
         """Extract subtitle streams one ffmpeg invocation at a time (to stdout).
 
         This preserves the original extraction behavior and is used as a
         fallback whenever single-pass extraction is unavailable (no ffprobe) or
-        fails. Stops at the first stream ffmpeg cannot extract.
+        fails. Stops at the first stream ffmpeg cannot extract. Returns
+        ``(stream, buffer)`` pairs.
         """
-        buffers: List[io.BytesIO] = []
+        buffers: List[Tuple[str, io.BytesIO]] = []
         for stream in streams:
             ffmpeg_args = [
                 ffmpeg_bin_path(
@@ -482,7 +494,7 @@ def _extract_embedded_subs_per_stream(
             output = process.communicate()[0]
             if process.returncode != 0:
                 break
-            buffers.append(io.BytesIO(output))
+            buffers.append((stream, io.BytesIO(output)))
         return buffers
 
     def try_fit_using_embedded_subs(self, fname: str) -> None:
@@ -495,7 +507,21 @@ def try_fit_using_embedded_subs(self, fname: str) -> None:
             # enumerate the subtitle streams so they can all be extracted in a
             # single ffmpeg pass (~5x faster than one invocation per stream)
             streams = self._probe_embedded_subtitle_streams(fname)
-            if streams:
+            if streams is None:
+                # ffprobe unavailable/failed: fall back to probing the first 5
+                # streams individually (should cover 99% of movies)
+                subtitle_buffers = self._extract_embedded_subs_per_stream(
+                    fname, list(map("0:s:{}".format, range(5)))
+                )
+            elif not streams:
+                # ffprobe worked and every subtitle stream is bitmap or forced;
+                # do not probe blindly, that would pick the skipped streams up
+                # again
+                raise ValueError(
+                    "Video file appears to lack a usable text subtitle stream "
+                    "(none, bitmap-only, or forced-only)"
+                )
+            else:
                 subtitle_buffers = self._extract_embedded_subs_single_pass(
                     fname, streams
                 )
@@ -505,30 +531,44 @@ def try_fit_using_embedded_subs(self, fname: str) -> None:
                     subtitle_buffers = self._extract_embedded_subs_per_stream(
                         fname, streams
                     )
-            else:
-                # ffprobe unavailable/failed: fall back to probing the first 5
-                # streams individually (should cover 99% of movies)
-                subtitle_buffers = self._extract_embedded_subs_per_stream(
-                    fname, list(map("0:s:{}".format, range(5)))
-                )
-        embedded_subs = []
-        embedded_subs_times = []
-        for buffer in subtitle_buffers:
+        # (stream, cue count, last cue time, speech step) per extracted stream
+        embedded_subs: List[Tuple[str, int, float, SubtitleSpeechTransformer]] = []
+        for stream, buffer in subtitle_buffers:
             pipe = cast(
                 Pipeline,
                 make_subtitle_speech_pipeline(start_seconds=self.start_seconds),
             ).fit(buffer)
             speech_step = pipe.steps[-1][1]
-            embedded_subs.append(speech_step)
-            embedded_subs_times.append(speech_step.max_time_)
+            num_cues = len(pipe.named_steps["parse"].subs_)
+            embedded_subs.append(
+                (stream, num_cues, speech_step.max_time_, speech_step)
+            )
         if len(embedded_subs) == 0:
             if self.ref_stream is None:
                 error_msg = "Video file appears to lack subtitle stream"
             else:
                 error_msg = "Stream {} not found".format(self.ref_stream)
             raise ValueError(error_msg)
-        # use longest set of embedded subs
-        subs_to_use = embedded_subs[int(np.argmax(embedded_subs_times))]
+        # Use the stream with the most cues; break ties by last cue time. The
+        # previous rule (latest cue wins) let a forced-only stub with one cue
+        # near the end beat a full dialogue track.
+        stream, num_cues, _, subs_to_use = max(
+            embedded_subs, key=lambda entry: (entry[1], entry[2])
+        )
+        if self.ref_stream is None and num_cues < self.min_embedded_subtitle_cues:
+            # Too little speech to align against; raising here makes ``fit``
+            # fall through to the audio VAD, exactly as if no stream existed.
+            raise ValueError(
+                "best embedded subtitle stream {} has only {} cues "
+                "(< {}); falling back to audio".format(
+                    stream, num_cues, self.min_embedded_subtitle_cues
+                )
+            )
+        logger.info(
+            "using embedded subtitle stream %s (%d cues) as reference",
+            stream,
+            num_cues,
+        )
         self.video_speech_results_ = subs_to_use.subtitle_speech_results_
 
     def _build_ffmpeg_args(self, fname: str) -> List[str]:

tests/test_embedded_subs.py

@@ -19,15 +19,18 @@
 from ffsubsync.speech_transformers import VideoSpeechTransformer
 
 
-def _srt(end_sec):
-    """A valid SRT string with one short cue per second up to ``end_sec``."""
+def _srt(end_sec, step=1):
+    """A valid SRT string with one short cue every ``step`` seconds up to ``end_sec``."""
     lines = []
-    for i in range(1, end_sec + 1):
-        lines.append(str(i))
+    for n, start in enumerate(range(0, end_sec, step), start=1):
+        lines.append(str(n))
         lines.append(
-            "00:00:{:02d},000 --> 00:00:{:02d},500".format(i - 1, i - 1)
+            "{:02d}:{:02d}:{:02d},000 --> {:02d}:{:02d}:{:02d},500".format(
+                start // 3600, start % 3600 // 60, start % 60,
+                start // 3600, start % 3600 // 60, start % 60,
+            )
         )
-        lines.append("line {}".format(i))
+        lines.append("line {}".format(n))
         lines.append("")
     return "\n".join(lines).encode("utf-8")
 
@@ -46,6 +49,9 @@ def _transformer(**overrides):
         sample_rate=SAMPLE_RATE,
         frame_rate=DEFAULT_FRAME_RATE,
         non_speech_label=0.0,
+        # the fixtures here are a few cues long; the sparse-stream threshold
+        # has its own tests below
+        min_embedded_subtitle_cues=0,
     )
     kwargs.update(overrides)
     return VideoSpeechTransformer(**kwargs)
@@ -217,11 +223,11 @@ def test_explicit_ref_stream_uses_per_stream(monkeypatch):
 def test_raises_when_no_subtitle_streams(monkeypatch):
     _install_fake_popen(
         monkeypatch,
-        probe_lines=[],  # ffprobe finds nothing -> fall back -> per-stream all fail
+        probe_lines=[],  # ffprobe finds nothing -> no usable stream, no ffmpeg
         srt_by_stream={},
     )
     t = _transformer()
-    with pytest.raises(ValueError, match="lack subtitle stream"):
+    with pytest.raises(ValueError, match="lack a usable text subtitle stream"):
         t.try_fit_using_embedded_subs("movie.mkv")
 
 
@@ -276,3 +282,84 @@ def test_build_ffmpeg_args_keeps_aresample_filter():
     args = t._build_ffmpeg_args("movie.mkv")
     assert "-af" in args
     assert args[args.index("-af") + 1] == "aresample=async=1"
+
+
+def test_forced_streams_are_skipped(monkeypatch):
+    # ffprobe now reports the forced disposition as a third column; a forced
+    # stream (signs / foreign lines only) must never be mapped as the reference
+    calls = _install_fake_popen(
+        monkeypatch,
+        probe_lines=["2,subrip,1", "3,subrip,0"],
+        srt_by_stream={"0:2": _srt(3000, step=10), "0:3": _srt(60)},
+    )
+    t = _transformer()
+    t.try_fit_using_embedded_subs("movie.mkv")
+
+    ffmpeg_call = next(
+        c for c in calls if "ffmpeg" in os.path.basename(c[0]).lower()
+    )
+    mapped = [ffmpeg_call[i + 1] for i, a in enumerate(ffmpeg_call) if a == "-map"]
+    assert mapped == ["0:3"]
+    assert np.array_equal(
+        t.video_speech_results_, _parse_speech_results(_srt(60))
+    )
+
+
+def test_stream_with_most_cues_wins_over_latest_cue(monkeypatch):
+    # 0:2 is a 3-cue stub whose last cue sits near the end of the film; 0:3 is a
+    # full dialogue track that ends earlier. The old rule (latest cue wins)
+    # picked the stub.
+    stub = _srt(7200, step=2400)  # cues at 0, 40 and 80 minutes
+    full = _srt(1200)  # 1200 cues, ends at 20 minutes
+    _install_fake_popen(
+        monkeypatch,
+        probe_lines=["2,subrip,0", "3,subrip,0"],
+        srt_by_stream={"0:2": stub, "0:3": full},
+    )
+    t = _transformer()
+    t.try_fit_using_embedded_subs("movie.mkv")
+    assert np.array_equal(t.video_speech_results_, _parse_speech_results(full))
+
+
+def test_sparse_streams_fall_back_to_audio(monkeypatch):
+    # with the default threshold a forced-style stub is treated as "no usable
+    # subtitle stream", which is the signal fit() uses to fall back to audio
+    _install_fake_popen(
+        monkeypatch,
+        probe_lines=["2,subrip,0"],
+        srt_by_stream={"0:2": _srt(40)},  # 40 cues < default 50
+    )
+    t = _transformer(min_embedded_subtitle_cues=st.DEFAULT_MIN_EMBEDDED_SUBTITLE_CUES)
+    with pytest.raises(ValueError, match="only 40 cues"):
+        t.try_fit_using_embedded_subs("movie.mkv")
+    assert t.video_speech_results_ is None
+
+
+def test_explicit_ref_stream_bypasses_sparse_threshold(monkeypatch):
+    # a stream the user asked for by name is used no matter how few cues it has
+    _install_fake_popen(
+        monkeypatch,
+        probe_lines=[],
+        srt_by_stream={"0:s:1": _srt(5)},
+    )
+    t = _transformer(
+        ref_stream="0:s:1",
+        min_embedded_subtitle_cues=st.DEFAULT_MIN_EMBEDDED_SUBTITLE_CUES,
+    )
+    t.try_fit_using_embedded_subs("movie.mkv")
+    assert np.array_equal(t.video_speech_results_, _parse_speech_results(_srt(5)))
+
+
+def test_only_forced_streams_means_no_usable_stream(monkeypatch):
+    # when ffprobe finds subtitle streams but every one is forced (or bitmap),
+    # the transformer must not fall back to blind per-stream probing, which
+    # would pick the forced stream up again; it raises so fit() uses audio
+    calls = _install_fake_popen(
+        monkeypatch,
+        probe_lines=["2,subrip,1", "3,hdmv_pgs_subtitle,0"],
+        srt_by_stream={"0:2": _srt(5), "0:s:0": _srt(5)},
+    )
+    t = _transformer()
+    with pytest.raises(ValueError, match="lack a usable text subtitle stream"):
+        t.try_fit_using_embedded_subs("movie.mkv")
+    assert _count_ffmpeg_invocations(calls) == 0