"""Subtitles synced to a LIVE stream -- one with no copy to measure (2026-10-18).
The library sync measures a track against the copy's own audio. A stream
being played from vidcore or HdHub has no copy yet, so Kodi got wyzie's raw
tracks, cut for some other release. Here the server takes six 150 s audio
windows of the very stream being played (vidcore: the 480p variant of the
same master, segments fetched directly because ffmpeg cannot seek its muxed
fMP4; HdHub: a ranged seek into the file, ENGLISH audio), and window_align
places each English track against them.
GET /subs/<kind>/<ident>.json?src=vidcore|hdhub|mp4&h=<height>
not held -> {"status": "pending"|"ready"|"none", "subtitles": [...]}
held -> the library answer, src ignored (the copy is what plays)
Only tracks the windows could place are offered: in sync as fetched, or
corrected. Anything else is not offered and the player keeps what it had.
A track without `src` behaves exactly as before, so an older Jacktook is
unaffected. Nothing here may start an ingest.
Run: .venv/bin/python3 -m pytest test_live_subsync.py -q
"""
import os
import numpy as np
import pytest
import addon
import window_align as wa
TMDB = {
"tv/1": {"name": "Show", "first_air_date": "2020-01-01", "seasons": []},
"tv/1/season/2/episode/6": {"name": "Six", "runtime": 30},
"movie/2": {"title": "Film", "release_date": "2021-05-05", "runtime": 120},
}
@pytest.fixture(autouse=True)
def isolated(tmp_path, monkeypatch):
monkeypatch.setattr(addon, "LIBRARY_ROOT", str(tmp_path / "lib"))
monkeypatch.setattr(addon, "INGEST_QUEUE_FILE", str(tmp_path / "queue.json"))
monkeypatch.setattr(addon, "LIBRARY_VERIFIED_FILE", str(tmp_path / "verified.json"))
monkeypatch.setattr(addon, "MP4_DIR", str(tmp_path / "mp4-cache"))
monkeypatch.setattr(addon, "_duration_cache", {})
monkeypatch.setattr(addon, "_library_meta_cache", {})
monkeypatch.setattr(addon, "_mp4_jobs", {})
monkeypatch.setattr(addon, "_live_jobs", {})
monkeypatch.setattr(addon, "_tmdb_get", lambda path, params=None: TMDB[path])
monkeypatch.setattr(addon, "_media_duration", lambda p: 7200.0)
def refuse(*a, **k):
raise AssertionError("a subtitle request started a download")
monkeypatch.setattr(addon, "_start_mp4_job", refuse)
monkeypatch.setattr(addon, "_queue_next_episode", refuse)
monkeypatch.setattr(addon, "_ensure_ingest", refuse)
yield
@pytest.fixture
def client():
return addon.app.test_client()
# ---------------------------------------------------------- a fake stream
def _cues(total=7200.0):
rng = np.random.default_rng(3)
t, out = 3.0, []
while t < total:
out.append((t, t + 2.0 + rng.uniform(0, 1.5)))
t += 7.0 + rng.uniform(-2.5, 2.5)
return out
CUES = _cues()
POSITIONS = [720.0, 1800.0, 2880.0, 3960.0, 5040.0, 6120.0]
def _cut(t):
"""A different edit: scenes added and removed, so the shift changes along
the film and no single offset+scale repairs it."""
for edge, shift in ((1500.0, 0.0), (3500.0, -11.0), (5000.0, 23.0)):
if t < edge:
return shift
return -37.0
def _srt(cues, scale=1.0, offset=0.0):
out = []
for n, (a, b) in enumerate(cues, 1):
out.append(f"{n}\n{wa._fmt(a * scale + offset)} --> {wa._fmt(b * scale + offset)}\nline {n}\n")
return "\n".join(out)
def _windows():
"""Speech exactly where CUES are: the stream's truth."""
wins = []
for p in POSITIONS:
w = np.zeros(15000)
for a, b in CUES:
i, j = int(round((a - p) * 100)), int(round((b - p) * 100))
i, j = max(i, 0), min(j, 15000)
if i < j:
w[i:j] = 1.0
wins.append((p, w))
return wins
@pytest.fixture
def stream(monkeypatch):
"""Source, windows and English tracks stubbed; the sync runs inline."""
state = {"started": [], "tracks": [
("en", _srt(CUES, offset=-7.0)), # needs +7.0 -> corrected
("en.2", _srt(CUES)), # already right -> in sync
("en.3", _srt([(a + _cut(a), b + _cut(a)) for a, b in CUES])), # another cut
], "source": {"kind": "hls", "url": "http://cdn.invalid/v.m3u8"}}
monkeypatch.setattr(addon, "_live_source",
lambda *a, **k: state["source"])
monkeypatch.setattr(addon, "_live_windows", lambda source, work: _windows())
monkeypatch.setattr(addon, "_live_tracks", lambda *a, **k: state["tracks"])
def inline(key, target):
state["started"].append(key)
target()
monkeypatch.setattr(addon, "_start_live_sync", inline)
return state
def _hold(kind, season=0, episode=0, tmdb="2"):
meta = addon._library_meta(kind, f"tmdb:{tmdb}", season, episode)
path = addon._library_path(kind, meta["title"], meta["year"], meta["tmdb_id"],
season, episode, meta["episode_title"])
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "wb") as f:
f.write(b"0" * 1000)
with open(os.path.splitext(path)[0] + ".en.srt", "w") as f:
f.write("library")
return path
# ------------------------------------------------------------------ route
def test_without_src_the_answer_is_unchanged(client, stream):
r = client.get("/subs/movie/tmdb:2.json")
assert r.get_json() == {"subtitles": []}
assert stream["started"] == []
def test_a_live_request_runs_one_sync(client, stream):
client.get("/subs/movie/tmdb:2.json?src=vidcore&h=1080")
client.get("/subs/movie/tmdb:2.json?src=vidcore&h=1080")
assert len(stream["started"]) == 1
def test_pending_while_the_sync_runs(client, monkeypatch, stream):
monkeypatch.setattr(addon, "_start_live_sync", lambda key, target: None)
r = client.get("/subs/movie/tmdb:2.json?src=vidcore&h=1080").get_json()
assert r == {"status": "pending", "subtitles": []}
def test_ready_offers_only_the_tracks_the_windows_placed(client, stream):
r = client.get("/subs/movie/tmdb:2.json?src=vidcore&h=1080").get_json()
assert r["status"] == "ready"
by_id = {s["id"]: s for s in r["subtitles"]}
assert set(by_id) == {"en", "en.2"}, "offered a track from another cut"
assert by_id["en"]["status"] == "corrected"
assert by_id["en"]["offset"] == pytest.approx(7.0, abs=0.05)
assert by_id["en.2"]["status"] == "in_sync"
assert all(s["lang"] == "en" for s in r["subtitles"])
def test_a_corrected_track_is_served_moved(client, stream):
r = client.get("/subs/movie/tmdb:2.json?src=vidcore&h=1080").get_json()
url = next(s["url"] for s in r["subtitles"] if s["id"] == "en")
body = client.get(url.split("localhost", 1)[1]).get_data(as_text=True)
# Not the first cue: shifted -7 s, it was clamped to 0 when written.
assert wa.srt_cues(body)[5][0] == pytest.approx(CUES[5][0], abs=0.05)
def test_an_episode_is_its_own_sync(client, stream):
client.get("/subs/movie/tmdb:2.json?src=vidcore&h=1080")
client.get("/subs/series/tmdb:1:2:6.json?src=vidcore&h=1080")
assert len(stream["started"]) == 2
def test_each_source_is_its_own_sync(client, stream):
"""vidcore and HdHub are different releases: what fits one need not fit
the other."""
client.get("/subs/movie/tmdb:2.json?src=vidcore&h=1080")
client.get("/subs/movie/tmdb:2.json?src=hdhub&h=1080")
assert len(stream["started"]) == 2
def test_nothing_placeable_is_none(client, stream):
stream["tracks"] = stream["tracks"][2:]
r = client.get("/subs/movie/tmdb:2.json?src=vidcore&h=1080").get_json()
assert r == {"status": "none", "subtitles": []}
def test_no_source_is_none(client, stream):
stream["source"] = None
r = client.get("/subs/movie/tmdb:2.json?src=vidcore&h=1080").get_json()
assert r == {"status": "none", "subtitles": []}
def test_a_held_copy_answers_from_the_library_whatever_src_says(client, stream):
_hold("movie")
r = client.get("/subs/movie/tmdb:2.json?src=vidcore&h=1080").get_json()
assert [s["id"] for s in r["subtitles"]] == ["en"]
assert stream["started"] == []
def test_a_finished_sync_is_kept_on_disk(client, stream, monkeypatch):
client.get("/subs/movie/tmdb:2.json?src=vidcore&h=1080")
monkeypatch.setattr(addon, "_live_jobs", {}) # a restart
r = client.get("/subs/movie/tmdb:2.json?src=vidcore&h=1080").get_json()
assert r["status"] == "ready" and len(stream["started"]) == 1
def test_an_unknown_src_or_track_is_404(client, stream):
assert client.get("/subs/movie/tmdb:2.json?src=elsewhere").status_code == 404
client.get("/subs/movie/tmdb:2.json?src=vidcore&h=1080")
assert client.get("/subs/movie/tmdb:2/en.3.srt?src=vidcore&h=1080").status_code == 404
assert client.get("/subs/movie/tmdb:2/..%2Fstate.srt?src=vidcore&h=1080").status_code == 404
# --------------------------------------------------------- the HLS windows
PLAYLIST = """#EXTM3U
#EXT-X-TARGETDURATION:6
#EXT-X-PLAYLIST-TYPE:VOD
#EXT-X-MAP:URI="https://seg.invalid/init.mp4"
""" + "".join(f"#EXTINF:6.0,\nhttps://seg.invalid/s{i}.m4s\n" for i in range(1200)) \
+ "#EXT-X-ENDLIST\n"
def test_the_playlist_gives_init_segments_and_their_starts():
init, segs = addon._hls_segments(PLAYLIST, "https://cdn.invalid/v/index.m3u8")
assert init == "https://seg.invalid/init.mp4"
assert len(segs) == 1200
assert segs[0] == (0.0, "https://seg.invalid/s0.m4s")
assert segs[10] == (60.0, "https://seg.invalid/s10.m4s")
def test_each_window_takes_the_segments_that_cover_it():
_, segs = addon._hls_segments(PLAYLIST, "https://cdn.invalid/v/index.m3u8")
plan = addon._window_plan(segs, fractions=(0.1, 0.5), seconds=150)
assert [p[0] for p in plan] == [120, 600] # segment indexes
assert all(p[1] - p[0] == 25 for p in plan) # 150 s of 6 s segments
# -------------------------------------------------------- the English tracks
def test_live_tracks_are_wyzie_english(monkeypatch):
monkeypatch.setattr(addon, "_wyzie_tracks", lambda *a, **k: [
{"language": "en", "url": "u1"}, {"language": "fr", "url": "u2"},
{"language": "en", "url": "u3"}])
monkeypatch.setattr(addon, "_fetch_subtitle", lambda url: f"body {url}".encode())
monkeypatch.setattr(addon, "_opensubs_for",
lambda *a, **k: pytest.fail("asked OpenSubtitles"))
got = addon._live_tracks("movie", "tmdb:2", 0, 0)
assert got == [("en", "body u1"), ("en.2", "body u3")]
def test_live_tracks_fall_back_to_opensubtitles(monkeypatch):
monkeypatch.setattr(addon, "_wyzie_tracks", lambda *a, **k: [
{"language": "en", "url": "u1"}])
def wyzie_503(url):
raise RuntimeError("503")
monkeypatch.setattr(addon, "_fetch_subtitle",
lambda url: wyzie_503(url) if url == "u1" else b"os")
monkeypatch.setattr(addon, "_opensubs_for",
lambda kind, tmdb, s=0, e=0: [{"language": "en", "url": "o1"}])
assert addon._live_tracks("movie", "tmdb:2", 0, 0) == [("en", "os")]