pedro-bento
ed8c0145ff
All checks were successful
CI / lint-and-test (push) Successful in 11s
Reviewed-on: #1 Co-authored-by: pedro-bento <mail@pbento.pt> Co-committed-by: pedro-bento <mail@pbento.pt>
290 lines
8.9 KiB
Python
290 lines
8.9 KiB
Python
import json
|
|
import sys
|
|
|
|
import pytest
|
|
|
|
from cadence import (
|
|
Segment,
|
|
Word,
|
|
_cache_key,
|
|
_read_cache,
|
|
_write_cache,
|
|
build_timeline,
|
|
compute_jl_cut_shifts,
|
|
frames_to_tc,
|
|
map_words_to_edited_timeline,
|
|
normalize_token,
|
|
parse_args,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# normalize_token
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.parametrize(
|
|
("raw", "expected"),
|
|
[
|
|
("Uh...", "uh"),
|
|
("—", ""),
|
|
('"Hello,"', "hello"),
|
|
(" Word? ", "word"),
|
|
("...", ""),
|
|
("Umm!", "umm"),
|
|
],
|
|
)
|
|
def test_normalize_token(raw, expected):
|
|
assert normalize_token(raw) == expected
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# frames_to_tc
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_frames_to_tc_zero_and_negative():
|
|
assert frames_to_tc(0, 30.0) == "00:00:00.000"
|
|
assert frames_to_tc(-7, 30.0) == "00:00:00.000"
|
|
|
|
|
|
def _parse_tc(tc: str) -> float:
|
|
h, m, s = tc.split(":")
|
|
return int(h) * 3600 + int(m) * 60 + float(s)
|
|
|
|
|
|
@pytest.mark.parametrize("frames", [1, 30, 100, 1000, 30000, 90000])
|
|
def test_frames_to_tc_roundtrip_ntsc(frames):
|
|
fps = 30000 / 1001
|
|
tc = frames_to_tc(frames, fps)
|
|
secs = _parse_tc(tc)
|
|
back = round(secs * fps)
|
|
assert abs(back - frames) <= 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# build_timeline
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _assert_tiling(timeline, duration, min_keep):
|
|
assert timeline, "timeline must not be empty"
|
|
|
|
# Alternating keep/drop actions.
|
|
for prev, nxt in zip(timeline, timeline[1:]):
|
|
assert prev.action != nxt.action
|
|
|
|
# Contiguous tiling of [0, duration].
|
|
assert timeline[0].start == pytest.approx(0.0)
|
|
for prev, nxt in zip(timeline, timeline[1:]):
|
|
assert prev.end == pytest.approx(nxt.start)
|
|
assert timeline[-1].end == pytest.approx(duration)
|
|
|
|
keep = sum(s.end - s.start for s in timeline if s.action == "keep")
|
|
drop = sum(s.end - s.start for s in timeline if s.action == "drop")
|
|
assert keep == pytest.approx(duration - drop)
|
|
|
|
for seg in timeline:
|
|
if seg.action == "keep":
|
|
assert (seg.end - seg.start) >= min_keep - 1e-9
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("cuts", "duration"),
|
|
[
|
|
([], 10.0),
|
|
([(2.0, 3.0, "silence"), (5.0, 6.0, "filler")], 10.0),
|
|
([(0.0, 1.0, "x"), (9.5, 10.0, "y")], 10.0),
|
|
([(0.0, 2.0, "a"), (2.02, 4.0, "b")], 5.0),
|
|
([(1.0, 2.0, "a"), (2.0, 3.0, "b"), (3.0, 4.0, "c")], 4.0),
|
|
],
|
|
)
|
|
def test_build_timeline_invariants(cuts, duration):
|
|
min_keep = 0.08
|
|
timeline = build_timeline(cuts, duration, min_keep_dur=min_keep)
|
|
_assert_tiling(timeline, duration, min_keep)
|
|
|
|
|
|
def test_build_timeline_no_cuts_single_keep():
|
|
assert build_timeline([], 5.0) == [Segment(0.0, 5.0, "keep")]
|
|
|
|
|
|
def test_build_timeline_zero_duration_returns_keep():
|
|
# Verified real behavior: the empty list falls through to a single keep
|
|
# segment spanning [0, 0].
|
|
assert build_timeline([], 0.0) == [Segment(0.0, 0.0, "keep")]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# compute_jl_cut_shifts
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _keeps(*pairs):
|
|
return [Segment(s, e, "keep") for s, e in pairs]
|
|
|
|
|
|
def _assert_jl_bounds(segs, shifts, fps, jl_frames):
|
|
for i in range(len(segs) - 1):
|
|
cur_in = int(round(segs[i].start * fps))
|
|
cur_out = int(round(segs[i].end * fps))
|
|
cur_dur = max(1, cur_out - cur_in)
|
|
next_in = int(round(segs[i + 1].start * fps))
|
|
next_out = int(round(segs[i + 1].end * fps))
|
|
next_dur = max(1, next_out - next_in)
|
|
handle = max(0, next_in - cur_out - 1)
|
|
bound = min(jl_frames, handle, cur_dur // 4, next_dur // 4)
|
|
assert abs(shifts[i]) <= bound
|
|
assert shifts[-1] == 0
|
|
|
|
|
|
def test_jl_off_and_trivial_cases():
|
|
segs = _keeps((0, 1), (2, 3), (4, 5))
|
|
assert compute_jl_cut_shifts(segs, 30.0, "off", 10) == [0, 0, 0]
|
|
assert compute_jl_cut_shifts(segs, 30.0, "j", 0) == [0, 0, 0]
|
|
assert compute_jl_cut_shifts(_keeps((0, 1)), 30.0, "j", 10) == [0]
|
|
assert compute_jl_cut_shifts([], 30.0, "j", 10) == []
|
|
|
|
|
|
def test_jl_j_signs_and_bounds():
|
|
segs = _keeps((0, 1), (2, 3), (4, 5))
|
|
shifts = compute_jl_cut_shifts(segs, 30.0, "j", 2)
|
|
assert shifts == [2, 2, 0]
|
|
assert all(x >= 0 for x in shifts)
|
|
_assert_jl_bounds(segs, shifts, 30.0, 2)
|
|
|
|
|
|
def test_jl_l_signs_and_bounds():
|
|
segs = _keeps((0, 1), (2, 3), (4, 5))
|
|
shifts = compute_jl_cut_shifts(segs, 30.0, "l", 2)
|
|
assert shifts == [-2, -2, 0]
|
|
assert all(x <= 0 for x in shifts)
|
|
_assert_jl_bounds(segs, shifts, 30.0, 2)
|
|
|
|
|
|
def test_jl_clamped_by_handle():
|
|
# Touching segments leave no handle, so even a large request becomes 0.
|
|
segs = _keeps((0, 1), (1, 2))
|
|
assert compute_jl_cut_shifts(segs, 30.0, "j", 100) == [0, 0]
|
|
|
|
|
|
def test_jl_last_element_always_zero():
|
|
segs = _keeps((0, 1), (2, 3), (4, 5), (6, 7))
|
|
for mode in ("j", "l"):
|
|
assert compute_jl_cut_shifts(segs, 30.0, mode, 2)[-1] == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# map_words_to_edited_timeline
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_map_words_to_edited_timeline(make_track):
|
|
td1 = make_track(
|
|
name="a.mp4",
|
|
duration=4.0,
|
|
timeline=[
|
|
Segment(0.0, 2.0, "keep"),
|
|
Segment(2.0, 3.0, "drop"),
|
|
Segment(3.0, 4.0, "keep"),
|
|
],
|
|
words=[
|
|
Word("a", 1.0, 1.2),
|
|
Word("dropped", 2.1, 2.3),
|
|
Word("b", 3.1, 3.3),
|
|
],
|
|
)
|
|
td2 = make_track(
|
|
name="b.mp4",
|
|
duration=3.0,
|
|
timeline=[Segment(0.0, 3.0, "keep")],
|
|
words=[Word("c", 0.5, 0.7)],
|
|
)
|
|
|
|
mapped = map_words_to_edited_timeline([td1, td2])
|
|
onsets = {text: (s, e) for s, e, text in mapped}
|
|
|
|
assert "dropped" not in onsets
|
|
assert onsets["b"][0] == pytest.approx(2.1)
|
|
assert onsets["c"][0] == pytest.approx(3.5)
|
|
assert len(mapped) == 3
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# cache helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_cache_roundtrip_and_no_version_key(tmp_path):
|
|
path = tmp_path / "cache.json"
|
|
words = [Word("Hello", 0.0, 0.5), Word("world", 0.5, 1.0)]
|
|
|
|
_write_cache(path, words)
|
|
|
|
assert _read_cache(path) == words
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
assert "version" not in payload
|
|
assert payload["words"][0] == {"text": "Hello", "start": 0.0, "end": 0.5}
|
|
|
|
|
|
def test_write_cache_empty_words_writes_nothing(tmp_path):
|
|
path = tmp_path / "empty.json"
|
|
_write_cache(path, [])
|
|
assert not path.exists()
|
|
|
|
|
|
def test_read_cache_corrupt_returns_none(tmp_path):
|
|
path = tmp_path / "bad.json"
|
|
path.write_text("{not valid json", encoding="utf-8")
|
|
assert _read_cache(path) is None
|
|
|
|
|
|
def test_read_cache_empty_words_returns_empty_list(tmp_path):
|
|
path = tmp_path / "empty.json"
|
|
path.write_text(json.dumps({"words": []}), encoding="utf-8")
|
|
assert _read_cache(path) == []
|
|
|
|
|
|
def test_read_cache_bare_list_returns_none(tmp_path):
|
|
path = tmp_path / "list.json"
|
|
path.write_text(json.dumps([{"text": "x", "start": 0, "end": 1}]), encoding="utf-8")
|
|
assert _read_cache(path) is None
|
|
|
|
|
|
def test_cache_key_varies_with_model_and_language():
|
|
base = _cache_key("large-v3", "faster-whisper", "en")
|
|
assert base != _cache_key("small", "faster-whisper", "en")
|
|
assert base != _cache_key("large-v3", "faster-whisper", "fr")
|
|
assert base == _cache_key("large-v3", "faster-whisper", "en")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# parse_args
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _parse(monkeypatch, argv):
|
|
monkeypatch.setattr(sys, "argv", ["cadence.py", *argv])
|
|
return parse_args()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"argv",
|
|
[
|
|
["video.mp4", "--audio-track", "0"],
|
|
["video.mp4", "--jl-frames", "-1"],
|
|
["video.mp4", "--pad", "-0.1"],
|
|
["video.mp4", "--max-silence", "abc"],
|
|
],
|
|
)
|
|
def test_parse_args_rejects_invalid(monkeypatch, argv):
|
|
with pytest.raises(SystemExit) as exc:
|
|
_parse(monkeypatch, argv)
|
|
assert exc.value.code != 0
|
|
|
|
|
|
def test_parse_args_defaults_are_none(monkeypatch):
|
|
args = _parse(monkeypatch, ["video.mp4"])
|
|
assert args.max_silence is None
|
|
assert args.pad is None
|
|
assert args.min_keep is None
|
|
assert args.jl_frames is None
|
|
assert args.preset is None
|
|
|
|
|
|
def test_parse_args_version_exits_zero(monkeypatch):
|
|
with pytest.raises(SystemExit) as exc:
|
|
_parse(monkeypatch, ["--version"])
|
|
assert exc.value.code == 0
|