OverviewFilesCommitsIssuesDocsReleasesPeers

stremio-vsembed commits / 2ea17d8

window_align: measure a subtitle against a few audio windows
For live streams, which have no local file and whose full audio costs as much as the download. Per-window lag at each candidate scale; believed only when enough good windows agree, else no answer. On real windows it recovers Hope live 0.16 s @1.000 (truth 0.13), No Other Choice en 7.05 @1.001 (7.01) and fr 0.36 @1.000 (0.33). ffsubsync's own search over the sparse reference said 59.19 @1.043 for Hope. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2ea17d8db16ca4d3d5aef3cea2bdc79e6d994ce7
by plankatron <fleeky@gmail.com> · 2026-10-19T00:56:12.000Z

Changed files (2) +242 -0

test_window_align.py +135 -0
……@@ -0,0 +1,135 @@
1"""Aligning a subtitle against a few audio WINDOWS of a stream (2026-10-18).
2
3A live stream has no local file to extract speech from, and taking all of
4its audio costs as much as the download (Hope: 767 s). So a handful of
5windows are fetched and each is measured on its own: at each candidate frame
6rate scale, the lag that best lines the subtitle up with that window's
7speech. A scale is believed only when enough good windows AGREE on one
8offset; otherwise there is no answer and the track is left as it was.
9
10ffsubsync's own search over a mostly-empty reference was tried first and
11failed on real data (Hope live: 59.19 s at 1.043, truth 0.13 s at 1.000) --
12one window of music the VAD called 97% speech was enough. Per-window lags
13with an agreement rule recovered all three known answers:
14 No Other Choice en 7.06/7.99/7.04 s @1.001 (full reference: 7.01 @1.001)
15 No Other Choice fr 0.37/1.32/0.36 s @1.000 (0.33 @1.000)
16 Hope live 0.17/0.11/0.16 s @1.000 (library copy: 0.13 @1.000)
17
18Run: .venv/bin/python3 -m pytest test_window_align.py -q
19"""
20
21import numpy as np
22import pytest
23
24import window_align as wa
25
26
27def _cues_every(seconds_total, period=7.0, length=2.5, start=3.0):
28 t, out = start, []
29 rng = np.random.default_rng(7)
30 while t < seconds_total:
31 out.append((t, t + length + rng.uniform(0, 1.5)))
32 t += period + rng.uniform(-2.5, 2.5)
33 return out
34
35
36def _speech_window(cues, pos, seconds, scale=1.0, offset=0.0):
37 """What VAD would hear in [pos, pos+seconds) if the subtitle needed
38 `scale * t + offset` to match: speech where the CORRECTED cues are."""
39 n = int(seconds * wa.SAMPLE_RATE)
40 w = np.zeros(n)
41 for a, b in cues:
42 i = int(round((a * scale + offset - pos) * wa.SAMPLE_RATE))
43 j = int(round((b * scale + offset - pos) * wa.SAMPLE_RATE))
44 i, j = max(i, 0), min(j, n)
45 if i < j:
46 w[i:j] = 1.0
47 return w
48
49
50FILM = 7200.0
51CUES = _cues_every(FILM)
52POSITIONS = [720.0, 1800.0, 2880.0, 3960.0, 5040.0, 6120.0]
53
54
55def _windows(scale=1.0, offset=0.0, seconds=150.0):
56 return [(p, _speech_window(CUES, p, seconds, scale, offset)) for p in POSITIONS]
57
58
59# ------------------------------------------------------------------ parsing
60
61def test_cues_are_read_from_srt():
62 srt = ("1\n00:00:01,500 --> 00:00:03,000\nhi\n\n"
63 "2\n01:02:03.250 --> 01:02:04,000\nthere\n")
64 assert wa.srt_cues(srt) == [(1.5, 3.0), (3723.25, 3724.0)]
65
66
67# ---------------------------------------------------------------- aligning
68
69def test_a_track_in_sync_measures_zero():
70 m = wa.align(CUES, _windows())
71 assert m["scale"] == 1.0 and abs(m["offset"]) <= 0.02
72
73
74def test_a_constant_offset_is_found():
75 m = wa.align(CUES, _windows(offset=7.0))
76 assert m["scale"] == 1.0 and m["offset"] == pytest.approx(7.0, abs=0.02)
77
78
79def test_frame_rate_drift_is_found_with_its_offset():
80 """No Other Choice's English: 24 vs 23.976 fps plus a shift."""
81 s = 24 / 23.976
82 m = wa.align(CUES, _windows(scale=s, offset=-2.0))
83 assert m["scale"] == pytest.approx(s) and m["offset"] == pytest.approx(-2.0, abs=0.05)
84
85
86def test_windows_the_vad_got_wrong_are_not_counted():
87 """Hope's window at 85% was music the VAD called 97% speech."""
88 wins = _windows(offset=3.0)
89 wins[2] = (wins[2][0], np.ones_like(wins[2][1]))
90 wins[4] = (wins[4][0], np.zeros_like(wins[4][1]))
91 m = wa.align(CUES, wins)
92 assert m["offset"] == pytest.approx(3.0, abs=0.02)
93 assert m["windows"] == 4
94
95
96def test_a_different_cut_gets_no_answer():
97 """Each window wants a different offset: no offset+scale repairs that,
98 and guessing one would be worse than leaving the track alone."""
99 wins = [(p, _speech_window(CUES, p, 150.0, 1.0, off))
100 for p, off in zip(POSITIONS, (0.0, 11.0, -23.0, 37.0, -5.0, 29.0))]
101 assert wa.align(CUES, wins) is None
102
103
104def test_one_good_window_is_not_enough():
105 wins = _windows(offset=3.0)[:1] + [(p, np.ones(15000)) for p in POSITIONS[1:]]
106 assert wa.align(CUES, wins) is None
107
108
109def test_no_windows_is_no_answer():
110 assert wa.align(CUES, []) is None
111 assert wa.align([], _windows()) is None
112
113
114# ---------------------------------------------------------------- applying
115
116def test_the_correction_is_applied_to_every_timestamp():
117 srt = ("1\n00:00:10,000 --> 00:00:12,500\nhi\n\n"
118 "2\n01:00:00,000 --> 01:00:02,000\nthere\n")
119 out = wa.apply(srt, scale=1.001, offset=-2.0)
120 assert "00:00:08,010 --> 00:00:10,512" in out
121 assert "01:00:01,600 --> 01:00:03,602" in out
122 assert "hi" in out and "there" in out
123
124
125def test_a_cue_pushed_before_zero_is_clamped():
126 out = wa.apply("1\n00:00:01,000 --> 00:00:02,000\nx\n", scale=1.0, offset=-1.5)
127 assert "00:00:00,000 --> 00:00:00,500" in out
128
129
130def test_a_corrected_track_measures_in_sync():
131 s = 24 / 23.976
132 m = wa.align(CUES, _windows(scale=s, offset=-2.0))
133 fixed = [(a * m["scale"] + m["offset"], b * m["scale"] + m["offset"]) for a, b in CUES]
134 again = wa.align(fixed, _windows(scale=s, offset=-2.0))
135 assert again["scale"] == 1.0 and abs(again["offset"]) <= 0.05
window_align.py +107 -0
……@@ -0,0 +1,107 @@
1"""Align a subtitle against a few audio windows of a stream.
2
3Pure functions, no I/O: addon.py fetches the windows and runs VAD on them
4(ffsubsync --serialize-speech), this decides what they say. See
5test_window_align.py for why ffsubsync's own search is not used here.
6
7Model (ffsubsync's, so results read the same as the library sync's):
8 corrected time = scale * t + offset
9"""
10
11import re
12from typing import Optional
13
14import numpy as np
15
16SAMPLE_RATE = 100 # ffsubsync's speech arrays are 100 Hz
17# 1.0 first: on a tie it wins, and an exact 1.0 is what "in sync" compares.
18SCALES = (1.0, 24 / 23.976, 23.976 / 24, 25 / 23.976, 23.976 / 25, 25 / 24, 24 / 25)
19MAX_LAG = 60.0 # seconds searched either side
20MIN_SCORE = 0.15 # measured: matching windows 0.17-0.60,
21 # wrong ones -0.16-0.09 (Hope, NOC)
22AGREE = 0.5 # seconds; NOC en windows spread 0.95
23 # only where the audio had a gap
24SPEECH_FRAC = (0.05, 0.9) # outside this the VAD heard music/nothing
25
26_TS = r"(\d+):(\d{2}):(\d{2})[,.](\d{1,3})"
27_CUE = re.compile(_TS + r"(\s*-->\s*)" + _TS)
28
29
30def _secs(h, m, s, ms) -> float:
31 return int(h) * 3600 + int(m) * 60 + int(s) + int(ms.ljust(3, "0")) / 1000
32
33
34def srt_cues(text: str) -> list:
35 """[(start, end)] in seconds, in file order."""
36 return [(_secs(*m.group(1, 2, 3, 4)), _secs(*m.group(6, 7, 8, 9)))
37 for m in _CUE.finditer(text)]
38
39
40def _signal(cues, scale: float, start: float, length: int) -> np.ndarray:
41 sig = np.zeros(length)
42 for a, b in cues:
43 i = int(round((a * scale - start) * SAMPLE_RATE))
44 j = int(round((b * scale - start) * SAMPLE_RATE))
45 i, j = max(i, 0), min(j, length)
46 if i < j:
47 sig[i:j] = 1.0
48 return sig
49
50
51def _lag(cues, scale: float, pos: float, speech: np.ndarray) -> tuple:
52 """(offset, score) that best lines the scaled cues up with one window.
53 Score is the ±1 agreement per sample, so -1..1 whatever the length."""
54 pad = int(MAX_LAG * SAMPLE_RATE)
55 n = len(speech)
56 w = 2 * speech - 1
57 s = 2 * _signal(cues, scale, pos - MAX_LAG, n + 2 * pad) - 1
58 size = 1 << int(np.ceil(np.log2(len(s) + n)))
59 corr = np.fft.irfft(np.fft.rfft(s, size) * np.conj(np.fft.rfft(w, size)), size)
60 valid = corr[:2 * pad + 1]
61 k = int(np.argmax(valid))
62 return (pad - k) / SAMPLE_RATE, float(valid[k]) / n
63
64
65def align(cues, windows) -> Optional[dict]:
66 """{"offset", "scale", "windows", "score"} or None when the windows do not
67 agree on an answer. `windows` is [(position_seconds, speech_array)]."""
68 if not cues:
69 return None
70 usable = []
71 for pos, speech in windows:
72 speech = (np.asarray(speech, dtype=float) >= 1.0).astype(float)
73 if len(speech) and SPEECH_FRAC[0] < speech.mean() < SPEECH_FRAC[1]:
74 usable.append((pos, speech))
75 best = None
76 for scale in SCALES:
77 good = [r for r in (_lag(cues, scale, p, w) for p, w in usable)
78 if r[1] >= MIN_SCORE]
79 if len(good) < 2:
80 continue
81 cluster = max(([g for g in good if abs(g[0] - o) <= AGREE] for o, _ in good),
82 key=lambda c: (len(c), sum(x[1] for x in c)))
83 if len(cluster) < 2 or 2 * len(cluster) < len(good):
84 continue
85 cand = {"offset": round(float(np.median([c[0] for c in cluster])), 2),
86 "scale": scale, "windows": len(cluster),
87 "score": round(sum(c[1] for c in cluster) / len(cluster), 3)}
88 if best is None or (cand["windows"], cand["score"]) > (best["windows"], best["score"]):
89 best = cand
90 return best
91
92
93def _fmt(t: float) -> str:
94 ms = max(0, int(round(t * 1000)))
95 h, ms = divmod(ms, 3_600_000)
96 m, ms = divmod(ms, 60_000)
97 s, ms = divmod(ms, 1000)
98 return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}"
99
100
101def apply(text: str, scale: float, offset: float) -> str:
102 """The SRT with every timestamp moved to scale * t + offset (>= 0)."""
103 def one(m):
104 a = _secs(*m.group(1, 2, 3, 4)) * scale + offset
105 b = _secs(*m.group(6, 7, 8, 9)) * scale + offset
106 return _fmt(a) + m.group(5) + _fmt(b)
107 return _CUE.sub(one, text)