"""Aligning a subtitle against a few audio WINDOWS of a stream (2026-10-18).
A live stream has no local file to extract speech from, and taking all of
its audio costs as much as the download (Hope: 767 s). So a handful of
windows are fetched and each is measured on its own: at each candidate frame
rate scale, the lag that best lines the subtitle up with that window's
speech. A scale is believed only when enough good windows AGREE on one
offset; otherwise there is no answer and the track is left as it was.
ffsubsync's own search over a mostly-empty reference was tried first and
failed on real data (Hope live: 59.19 s at 1.043, truth 0.13 s at 1.000) --
one window of music the VAD called 97% speech was enough. Per-window lags
with an agreement rule recovered all three known answers:
No Other Choice en 7.06/7.99/7.04 s @1.001 (full reference: 7.01 @1.001)
No Other Choice fr 0.37/1.32/0.36 s @1.000 (0.33 @1.000)
Hope live 0.17/0.11/0.16 s @1.000 (library copy: 0.13 @1.000)
Run: .venv/bin/python3 -m pytest test_window_align.py -q
"""
import numpy as np
import pytest
import window_align as wa
def _cues_every(seconds_total, period=7.0, length=2.5, start=3.0):
t, out = start, []
rng = np.random.default_rng(7)
while t < seconds_total:
out.append((t, t + length + rng.uniform(0, 1.5)))
t += period + rng.uniform(-2.5, 2.5)
return out
def _speech_window(cues, pos, seconds, scale=1.0, offset=0.0):
"""What VAD would hear in [pos, pos+seconds) if the subtitle needed
`scale * t + offset` to match: speech where the CORRECTED cues are."""
n = int(seconds * wa.SAMPLE_RATE)
w = np.zeros(n)
for a, b in cues:
i = int(round((a * scale + offset - pos) * wa.SAMPLE_RATE))
j = int(round((b * scale + offset - pos) * wa.SAMPLE_RATE))
i, j = max(i, 0), min(j, n)
if i < j:
w[i:j] = 1.0
return w
FILM = 7200.0
CUES = _cues_every(FILM)
POSITIONS = [720.0, 1800.0, 2880.0, 3960.0, 5040.0, 6120.0]
def _windows(scale=1.0, offset=0.0, seconds=150.0):
return [(p, _speech_window(CUES, p, seconds, scale, offset)) for p in POSITIONS]
# ------------------------------------------------------------------ parsing
def test_cues_are_read_from_srt():
srt = ("1\n00:00:01,500 --> 00:00:03,000\nhi\n\n"
"2\n01:02:03.250 --> 01:02:04,000\nthere\n")
assert wa.srt_cues(srt) == [(1.5, 3.0), (3723.25, 3724.0)]
# ---------------------------------------------------------------- aligning
def test_a_track_in_sync_measures_zero():
m = wa.align(CUES, _windows())
assert m["scale"] == 1.0 and abs(m["offset"]) <= 0.02
def test_a_constant_offset_is_found():
m = wa.align(CUES, _windows(offset=7.0))
assert m["scale"] == 1.0 and m["offset"] == pytest.approx(7.0, abs=0.02)
def test_frame_rate_drift_is_found_with_its_offset():
"""No Other Choice's English: 24 vs 23.976 fps plus a shift."""
s = 24 / 23.976
m = wa.align(CUES, _windows(scale=s, offset=-2.0))
assert m["scale"] == pytest.approx(s) and m["offset"] == pytest.approx(-2.0, abs=0.05)
def test_windows_the_vad_got_wrong_are_not_counted():
"""Hope's window at 85% was music the VAD called 97% speech."""
wins = _windows(offset=3.0)
wins[2] = (wins[2][0], np.ones_like(wins[2][1]))
wins[4] = (wins[4][0], np.zeros_like(wins[4][1]))
m = wa.align(CUES, wins)
assert m["offset"] == pytest.approx(3.0, abs=0.02)
assert m["windows"] == 4
def test_a_different_cut_gets_no_answer():
"""Each window wants a different offset: no offset+scale repairs that,
and guessing one would be worse than leaving the track alone."""
wins = [(p, _speech_window(CUES, p, 150.0, 1.0, off))
for p, off in zip(POSITIONS, (0.0, 11.0, -23.0, 37.0, -5.0, 29.0))]
assert wa.align(CUES, wins) is None
def test_one_good_window_is_not_enough():
wins = _windows(offset=3.0)[:1] + [(p, np.ones(15000)) for p in POSITIONS[1:]]
assert wa.align(CUES, wins) is None
def test_no_windows_is_no_answer():
assert wa.align(CUES, []) is None
assert wa.align([], _windows()) is None
# ---------------------------------------------------------------- applying
def test_the_correction_is_applied_to_every_timestamp():
srt = ("1\n00:00:10,000 --> 00:00:12,500\nhi\n\n"
"2\n01:00:00,000 --> 01:00:02,000\nthere\n")
out = wa.apply(srt, scale=1.001, offset=-2.0)
assert "00:00:08,010 --> 00:00:10,512" in out
assert "01:00:01,600 --> 01:00:03,602" in out
assert "hi" in out and "there" in out
def test_a_cue_pushed_before_zero_is_clamped():
out = wa.apply("1\n00:00:01,000 --> 00:00:02,000\nx\n", scale=1.0, offset=-1.5)
assert "00:00:00,000 --> 00:00:00,500" in out
def test_a_corrected_track_measures_in_sync():
s = 24 / 23.976
m = wa.align(CUES, _windows(scale=s, offset=-2.0))
fixed = [(a * m["scale"] + m["offset"], b * m["scale"] + m["offset"]) for a, b in CUES]
again = wa.align(fixed, _windows(scale=s, offset=-2.0))
assert again["scale"] == 1.0 and abs(again["offset"]) <= 0.05