754 lines
28 KiB
Python
754 lines
28 KiB
Python
import tempfile, os, json
|
|
from main import build_export_path, format_time, build_ffmpeg_command, build_sequence_dir, build_audio_extract_command, resolve_keyframe, apply_keyframes_to_jobs
|
|
from core.ffmpeg import build_audio_clip_command, build_crossfade_merge_command, build_audio_delete_command, build_audio_silence_command, build_audio_reverse_command
|
|
from core.annotations import build_annotation_json_path, upsert_clip_annotation
|
|
from main import ProcessedDB
|
|
|
|
|
|
def test_build_export_path_first():
|
|
assert build_export_path("/out", "clip", 1) == "/out/clip_001.mp4"
|
|
|
|
def test_build_export_path_counter():
|
|
assert build_export_path("/out", "clip", 42) == "/out/clip_042.mp4"
|
|
|
|
def test_build_export_path_deep_counter():
|
|
assert build_export_path("/out", "shot", 999) == "/out/shot_999.mp4"
|
|
|
|
def test_build_export_path_sub():
|
|
assert build_export_path("/out", "clip", 1, sub=0) == "/out/clip_001_0.mp4"
|
|
assert build_export_path("/out", "clip", 1, sub=2) == "/out/clip_001_2.mp4"
|
|
|
|
def test_build_sequence_dir_sub():
|
|
assert build_sequence_dir("/out", "clip", 1, sub=0) == "/out/clip_001_0"
|
|
assert build_sequence_dir("/out", "clip", 1, sub=1) == "/out/clip_001_1"
|
|
|
|
def test_format_time_seconds():
|
|
assert format_time(0.0) == "0:00.0"
|
|
|
|
def test_format_time_minutes():
|
|
assert format_time(75.3) == "1:15.2"
|
|
|
|
def test_format_time_rounding():
|
|
assert format_time(61.05) == "1:01.0"
|
|
|
|
def test_format_time_no_sixty_rollover():
|
|
assert format_time(59.95) == "0:59.9"
|
|
|
|
|
|
def test_ffmpeg_command_no_resize():
|
|
cmd = build_ffmpeg_command("/in/video.mp4", 12.5, "/out/clip_001.mp4")
|
|
assert cmd[0] == "ffmpeg"
|
|
assert "-y" in cmd
|
|
assert "-ss" in cmd
|
|
assert str(12.5) in cmd
|
|
assert "-t" in cmd
|
|
assert "8" in cmd
|
|
assert cmd[-1] == "/out/clip_001.mp4"
|
|
assert "-vf" not in cmd
|
|
|
|
def test_ffmpeg_command_with_resize():
|
|
cmd = build_ffmpeg_command("/in/video.mp4", 0.0, "/out/clip_001.mp4", short_side=256)
|
|
assert "-vf" in cmd
|
|
vf_value = cmd[cmd.index("-vf") + 1]
|
|
assert "256" in vf_value
|
|
assert "scale" in vf_value
|
|
assert cmd[-1] == "/out/clip_001.mp4"
|
|
|
|
|
|
def test_audio_clip_command_exact_length():
|
|
cmd = build_audio_clip_command("/in/video.mp4", 12.5, 3.2, "/out/clip.wav")
|
|
assert cmd[0] == "ffmpeg"
|
|
# fast seek before input, exact duration, no video
|
|
assert cmd[cmd.index("-ss") + 1] == "12.5"
|
|
assert cmd[cmd.index("-t") + 1] == "3.2"
|
|
assert cmd.index("-ss") < cmd.index("-i")
|
|
assert "-vn" in cmd
|
|
assert cmd[-1] == "/out/clip.wav"
|
|
|
|
def test_audio_clip_command_codec_by_extension():
|
|
assert "pcm_s16le" in build_audio_clip_command("/in.mp4", 0, 1, "/o/a.wav")
|
|
assert "libmp3lame" in build_audio_clip_command("/in.mp4", 0, 1, "/o/a.mp3")
|
|
assert "flac" in build_audio_clip_command("/in.mp4", 0, 1, "/o/a.flac")
|
|
# Unknown extension -> no explicit -c:a, let ffmpeg pick from the container.
|
|
assert "-c:a" not in build_audio_clip_command("/in.mp4", 0, 1, "/o/a.xyz")
|
|
|
|
def test_audio_clip_command_extension_case_insensitive():
|
|
assert "flac" in build_audio_clip_command("/in.mp4", 0, 1, "/o/A.FLAC")
|
|
|
|
|
|
def test_audio_edit_filters_empty_when_defaults():
|
|
from core.ffmpeg import audio_edit_filters
|
|
assert audio_edit_filters(duration=3.0) == []
|
|
|
|
def test_audio_edit_filters_fade_normalize_gain():
|
|
from core.ffmpeg import audio_edit_filters
|
|
f = audio_edit_filters(duration=10.0, fade_in=0.5, fade_out=2.0,
|
|
normalize=True, gain_db=-3.0)
|
|
assert "afade=t=in:st=0:d=0.5" in f
|
|
assert "afade=t=out:st=8.0:d=2.0" in f # fade-out starts at duration - fade_out
|
|
assert "loudnorm" in f
|
|
assert "volume=-3.0dB" in f
|
|
|
|
def test_audio_clip_command_no_filters_unchanged():
|
|
cmd = build_audio_clip_command("/in.mp4", 1.0, 2.0, "/o/a.wav")
|
|
assert "-af" not in cmd
|
|
|
|
def test_audio_clip_command_appends_filter_chain():
|
|
cmd = build_audio_clip_command("/in.mp4", 1.0, 2.0, "/o/a.wav",
|
|
filters=["afade=t=in:st=0:d=0.5", "loudnorm"])
|
|
i = cmd.index("-af")
|
|
assert cmd[i + 1] == "afade=t=in:st=0:d=0.5,loudnorm"
|
|
assert i < len(cmd) - 1 and cmd[-1] == "/o/a.wav"
|
|
|
|
|
|
def test_merge_single_clip_reencodes():
|
|
cmd = build_crossfade_merge_command(["/a.wav"], 0.5, "/o/out.mp3")
|
|
assert cmd[0] == "ffmpeg"
|
|
assert cmd.count("-i") == 1
|
|
assert "libmp3lame" in cmd # codec by out ext
|
|
assert "acrossfade" not in " ".join(cmd)
|
|
assert "-map" not in cmd
|
|
assert "-filter_complex" not in cmd
|
|
assert cmd[-1] == "/o/out.mp3"
|
|
|
|
def test_merge_two_clips_acrossfade():
|
|
cmd = build_crossfade_merge_command(["/a.wav", "/b.wav"], 0.5, "/o/out.wav")
|
|
assert cmd.count("-i") == 2
|
|
fc = cmd[cmd.index("-filter_complex") + 1]
|
|
assert "[0][1]acrossfade=d=0.5" in fc
|
|
assert "[out]" in fc
|
|
assert cmd[cmd.index("-map") + 1] == "[out]"
|
|
assert "pcm_s16le" in cmd # codec by out ext (multi-clip path)
|
|
|
|
def test_merge_three_clips_chains():
|
|
cmd = build_crossfade_merge_command(["/a.wav", "/b.wav", "/c.wav"], 1.0, "/o/o.wav")
|
|
fc = cmd[cmd.index("-filter_complex") + 1]
|
|
assert fc.count("acrossfade=d=1.0") == 2 # two joins
|
|
assert "[0][1]acrossfade=d=1.0[a1]" in fc
|
|
assert "[a1][2]acrossfade=d=1.0[out]" in fc
|
|
assert cmd[cmd.index("-map") + 1] == "[out]"
|
|
|
|
def test_merge_zero_crossfade_uses_concat():
|
|
cmd = build_crossfade_merge_command(["/a.wav", "/b.wav"], 0.0, "/o/o.wav")
|
|
fc = cmd[cmd.index("-filter_complex") + 1]
|
|
assert "concat=n=2:v=0:a=1" in fc
|
|
assert "acrossfade" not in fc
|
|
|
|
def test_merge_empty_raises():
|
|
import pytest
|
|
with pytest.raises(ValueError):
|
|
build_crossfade_merge_command([], 0.5, "/o/o.wav")
|
|
|
|
def test_merge_per_join_crossfades():
|
|
cmd = build_crossfade_merge_command(["/a.wav", "/b.wav", "/c.wav"], [0.5, 1.5], "/o/o.wav")
|
|
fc = cmd[cmd.index("-filter_complex") + 1]
|
|
assert "[0][1]acrossfade=d=0.5[a1]" in fc
|
|
assert "[a1][2]acrossfade=d=1.5[out]" in fc
|
|
|
|
def test_merge_curve_emitted_when_non_default():
|
|
cmd = build_crossfade_merge_command(["/a.wav", "/b.wav"], 0.5, "/o/o.wav", curves="exp")
|
|
fc = cmd[cmd.index("-filter_complex") + 1]
|
|
assert "acrossfade=d=0.5:c1=exp:c2=exp[out]" in fc
|
|
|
|
def test_merge_default_curve_omits_c1c2():
|
|
cmd = build_crossfade_merge_command(["/a.wav", "/b.wav"], 0.5, "/o/o.wav") # tri default
|
|
fc = cmd[cmd.index("-filter_complex") + 1]
|
|
assert "c1=" not in fc # tri is ffmpeg's default -> omit
|
|
|
|
def test_merge_per_join_curves_list():
|
|
cmd = build_crossfade_merge_command(["/a.wav", "/b.wav", "/c.wav"], 0.5, "/o/o.wav",
|
|
curves=["tri", "exp"])
|
|
fc = cmd[cmd.index("-filter_complex") + 1]
|
|
assert "[0][1]acrossfade=d=0.5[a1]" in fc # tri -> no suffix
|
|
assert "acrossfade=d=0.5:c1=exp:c2=exp[out]" in fc # exp on 2nd join
|
|
|
|
def test_merge_mixed_zero_crossfade():
|
|
cmd = build_crossfade_merge_command(["/a.wav", "/b.wav", "/c.wav"], [0.0, 0.5], "/o/o.wav")
|
|
fc = cmd[cmd.index("-filter_complex") + 1]
|
|
assert "[0][1]concat=n=2:v=0:a=1[a1]" in fc # butt-join 1st
|
|
assert "acrossfade=d=0.5" in fc # crossfade 2nd
|
|
|
|
def test_merge_crossfade_list_length_mismatch_raises():
|
|
import pytest
|
|
with pytest.raises(ValueError):
|
|
build_crossfade_merge_command(["/a.wav", "/b.wav", "/c.wav"], [0.5], "/o/o.wav")
|
|
with pytest.raises(ValueError):
|
|
build_crossfade_merge_command(["/a.wav", "/b.wav"], [0.5, 0.5], "/o/o.wav")
|
|
|
|
def test_merge_curves_list_length_mismatch_raises():
|
|
import pytest
|
|
with pytest.raises(ValueError):
|
|
build_crossfade_merge_command(["/a.wav", "/b.wav", "/c.wav"], 0.5, "/o/o.wav", curves=["exp"])
|
|
|
|
def test_merge_unknown_curve_raises():
|
|
import pytest
|
|
with pytest.raises(ValueError):
|
|
build_crossfade_merge_command(["/a.wav", "/b.wav"], 0.5, "/o/o.wav", curves="bogus")
|
|
|
|
def test_merge_three_clips_zero_concat_chain():
|
|
cmd = build_crossfade_merge_command(["/a.wav", "/b.wav", "/c.wav"], 0.0, "/o/o.wav")
|
|
fc = cmd[cmd.index("-filter_complex") + 1]
|
|
assert fc.count("concat=n=2:v=0:a=1") == 2
|
|
assert "acrossfade" not in fc
|
|
|
|
|
|
# --- destructive audio region ops (delete/silence/reverse) ---
|
|
|
|
def test_audio_delete_command():
|
|
cmd = build_audio_delete_command("/in.wav", 1.0, 3.0, "/o/o.wav")
|
|
assert cmd[0] == "ffmpeg"
|
|
assert cmd.count("-i") == 1
|
|
fc = cmd[cmd.index("-filter_complex") + 1]
|
|
assert "atrim=end=1.0" in fc # keep [0,1]
|
|
assert "atrim=start=3.0" in fc # keep [3,end]
|
|
assert "concat=n=2:v=0:a=1[out]" in fc
|
|
assert cmd[cmd.index("-map") + 1] == "[out]"
|
|
assert "pcm_s16le" in cmd # codec by ext
|
|
assert cmd[-1] == "/o/o.wav"
|
|
|
|
def test_audio_silence_command():
|
|
cmd = build_audio_silence_command("/in.wav", 1.0, 3.0, "/o/o.mp3")
|
|
af = cmd[cmd.index("-af") + 1]
|
|
# volume=0 gated to the region; commas inside between() are escaped for the filtergraph
|
|
assert af == "volume=0:enable='between(t\\,1.0\\,3.0)'"
|
|
assert "libmp3lame" in cmd
|
|
assert cmd[-1] == "/o/o.mp3"
|
|
|
|
def test_audio_reverse_command():
|
|
cmd = build_audio_reverse_command("/in.wav", 1.0, 3.0, "/o/o.wav")
|
|
fc = cmd[cmd.index("-filter_complex") + 1]
|
|
assert "atrim=end=1.0" in fc # head [0,1]
|
|
assert "atrim=start=1.0:end=3.0" in fc # middle [1,3]
|
|
assert "areverse" in fc # reversed middle
|
|
assert "atrim=start=3.0" in fc # tail [3,end]
|
|
assert "concat=n=3:v=0:a=1[out]" in fc
|
|
assert cmd[cmd.index("-map") + 1] == "[out]"
|
|
|
|
def test_audio_delete_empty_head():
|
|
# deleting from 0 still produces a valid 2-branch concat (head is empty but harmless)
|
|
cmd = build_audio_delete_command("/in.wav", 0.0, 2.0, "/o/o.wav")
|
|
fc = cmd[cmd.index("-filter_complex") + 1]
|
|
assert "atrim=end=0.0" in fc and "atrim=start=2.0" in fc
|
|
|
|
|
|
def test_audio_heal_delete_command_crossfades_join():
|
|
from core.ffmpeg import build_audio_heal_delete_command
|
|
cmd = build_audio_heal_delete_command("/in.wav", 2.0, 4.0, "/o/o.wav", crossfade=0.1)
|
|
assert cmd[0] == "ffmpeg"
|
|
assert cmd.count("-i") == 1
|
|
fc = cmd[cmd.index("-filter_complex") + 1]
|
|
assert "atrim=end=2.0" in fc
|
|
assert "atrim=start=4.0" in fc
|
|
assert "acrossfade=d=0.1:c1=qsin:c2=qsin[out]" in fc
|
|
assert cmd[cmd.index("-map") + 1] == "[out]"
|
|
assert "pcm_s16le" in cmd
|
|
assert cmd[-1] == "/o/o.wav"
|
|
|
|
|
|
def test_audio_heal_delete_command_auto_crossfade_clamped():
|
|
from core.ffmpeg import build_audio_heal_delete_command
|
|
cmd = build_audio_heal_delete_command("/in.wav", 10.0, 12.0, "/o/o.mp3")
|
|
fc = cmd[cmd.index("-filter_complex") + 1]
|
|
assert "acrossfade=d=0.25:c1=qsin:c2=qsin[out]" in fc
|
|
assert "libmp3lame" in cmd
|
|
|
|
|
|
def test_audio_heal_delete_command_near_start_shortens_crossfade():
|
|
from core.ffmpeg import build_audio_heal_delete_command
|
|
cmd = build_audio_heal_delete_command("/in.wav", 0.03, 1.0, "/o/o.wav")
|
|
fc = cmd[cmd.index("-filter_complex") + 1]
|
|
assert "acrossfade=d=0.03:c1=qsin:c2=qsin[out]" in fc
|
|
|
|
|
|
def test_audio_heal_delete_command_rejects_invalid_region():
|
|
import pytest
|
|
from core.ffmpeg import build_audio_heal_delete_command
|
|
with pytest.raises(ValueError):
|
|
build_audio_heal_delete_command("/in.wav", 3.0, 3.0, "/o/o.wav")
|
|
|
|
|
|
# --- ProcessedDB ---
|
|
|
|
def test_db_add_and_get_markers():
|
|
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
|
|
path = f.name
|
|
try:
|
|
db = ProcessedDB(path)
|
|
db.add("video.mp4", 12.5, "/out/clip_001.mp4")
|
|
markers = db.get_markers("video.mp4")
|
|
assert len(markers) == 1
|
|
assert markers[0][0] == 12.5
|
|
finally:
|
|
os.unlink(path)
|
|
|
|
def test_db_exact_match_only():
|
|
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
|
|
path = f.name
|
|
try:
|
|
db = ProcessedDB(path)
|
|
db.add("episode_s01e01_2160p.mkv", 0.0, "/out/ep_001.mp4")
|
|
# Different filename — no match even if similar
|
|
assert db.get_markers("episode_s01e01_1080p.mkv") == []
|
|
# Exact filename — match
|
|
assert len(db.get_markers("episode_s01e01_2160p.mkv")) == 1
|
|
finally:
|
|
os.unlink(path)
|
|
|
|
def test_db_no_match():
|
|
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
|
|
path = f.name
|
|
try:
|
|
db = ProcessedDB(path)
|
|
db.add("alpha.mp4", 0.0, "/out/alpha_001.mp4")
|
|
assert db.get_markers("completely_different.mp4") == []
|
|
finally:
|
|
os.unlink(path)
|
|
|
|
def test_db_disabled_survives_bad_path():
|
|
db = ProcessedDB("/no/such/directory/8cut.db")
|
|
db.add("x.mp4", 0.0, "/out/x_001.mp4") # must not raise
|
|
assert db.get_markers("x.mp4") == []
|
|
|
|
def test_db_get_markers_returns_sorted():
|
|
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
|
|
path = f.name
|
|
try:
|
|
db = ProcessedDB(path)
|
|
db.add("video.mp4", 30.0, "/out/clip_002.mp4")
|
|
db.add("video.mp4", 10.0, "/out/clip_001.mp4")
|
|
db.add("video.mp4", 50.0, "/out/clip_003.mp4")
|
|
markers = db.get_markers("video.mp4")
|
|
assert len(markers) == 3
|
|
assert markers[0] == (10.0, 1, "/out/clip_001.mp4")
|
|
assert markers[1] == (30.0, 2, "/out/clip_002.mp4")
|
|
assert markers[2] == (50.0, 3, "/out/clip_003.mp4")
|
|
finally:
|
|
os.unlink(path)
|
|
|
|
def test_db_get_markers_no_match():
|
|
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
|
|
path = f.name
|
|
try:
|
|
db = ProcessedDB(path)
|
|
markers = db.get_markers("nothing.mp4")
|
|
assert markers == []
|
|
finally:
|
|
os.unlink(path)
|
|
|
|
def test_db_get_markers_disabled():
|
|
db = ProcessedDB("/no/such/directory/8cut.db")
|
|
assert db.get_markers("x.mp4") == []
|
|
|
|
def test_ffmpeg_command_portrait_only():
|
|
cmd = build_ffmpeg_command(
|
|
"/in/video.mp4", 0.0, "/out/clip.mp4",
|
|
portrait_ratio="9:16", crop_center=0.5,
|
|
)
|
|
assert "-vf" in cmd
|
|
vf = cmd[cmd.index("-vf") + 1]
|
|
assert "crop" in vf
|
|
assert "9" in vf
|
|
assert "scale" not in vf
|
|
assert cmd[-1] == "/out/clip.mp4"
|
|
|
|
def test_ffmpeg_command_portrait_and_resize():
|
|
cmd = build_ffmpeg_command(
|
|
"/in/video.mp4", 0.0, "/out/clip.mp4",
|
|
short_side=256, portrait_ratio="9:16", crop_center=0.5,
|
|
)
|
|
assert "-vf" in cmd
|
|
vf = cmd[cmd.index("-vf") + 1]
|
|
assert "crop" in vf
|
|
assert "scale" in vf
|
|
assert vf.index("crop") < vf.index("scale")
|
|
assert cmd[-1] == "/out/clip.mp4"
|
|
|
|
def test_ffmpeg_command_portrait_off():
|
|
cmd = build_ffmpeg_command("/in/video.mp4", 0.0, "/out/clip.mp4")
|
|
assert "-vf" not in cmd
|
|
|
|
# --- build_audio_extract_command ---
|
|
|
|
def test_audio_extract_output_path():
|
|
cmd = build_audio_extract_command("/in/v.mp4", 0.0, "/out/clip_001")
|
|
assert cmd[-1] == "/out/clip_001.wav"
|
|
|
|
def test_audio_extract_no_video():
|
|
cmd = build_audio_extract_command("/in/v.mp4", 0.0, "/out/clip_001")
|
|
assert "-vn" in cmd
|
|
|
|
def test_audio_extract_lossless_codec():
|
|
cmd = build_audio_extract_command("/in/v.mp4", 0.0, "/out/clip_001")
|
|
assert "-c:a" in cmd
|
|
assert cmd[cmd.index("-c:a") + 1] == "pcm_s16le"
|
|
|
|
def test_audio_extract_timing():
|
|
cmd = build_audio_extract_command("/in/v.mp4", 12.5, "/out/clip_001")
|
|
assert "-ss" in cmd
|
|
assert cmd[cmd.index("-ss") + 1] == "12.5"
|
|
assert "-t" in cmd
|
|
assert cmd[cmd.index("-t") + 1] == "8"
|
|
|
|
|
|
def test_build_sequence_dir_basic():
|
|
assert build_sequence_dir("/out", "clip", 1) == "/out/clip_001"
|
|
|
|
def test_build_sequence_dir_counter():
|
|
assert build_sequence_dir("/out", "clip", 42) == "/out/clip_042"
|
|
|
|
def test_ffmpeg_command_image_sequence():
|
|
cmd = build_ffmpeg_command("/in/v.mp4", 0.0, "/out/seq_001", image_sequence=True)
|
|
assert "-c:v" in cmd
|
|
assert cmd[cmd.index("-c:v") + 1] == "libwebp"
|
|
assert "-quality" in cmd
|
|
assert cmd[-1] == "/out/seq_001/frame_%04d.webp"
|
|
|
|
def test_ffmpeg_command_image_sequence_with_resize():
|
|
cmd = build_ffmpeg_command("/in/v.mp4", 0.0, "/out/seq_001", image_sequence=True, short_side=256)
|
|
assert "-vf" in cmd
|
|
vf = cmd[cmd.index("-vf") + 1]
|
|
assert "scale" in vf
|
|
assert cmd[-1] == "/out/seq_001/frame_%04d.webp"
|
|
|
|
def test_ffmpeg_command_image_sequence_no_audio():
|
|
cmd = build_ffmpeg_command("/in/v.mp4", 0.0, "/out/seq_001", image_sequence=True)
|
|
assert "-an" in cmd
|
|
assert "-c:a" not in cmd
|
|
assert "aac" not in cmd
|
|
|
|
|
|
def test_annotation_json_path():
|
|
assert build_annotation_json_path("/out") == "/out/dataset.json"
|
|
|
|
def test_upsert_creates_file():
|
|
with tempfile.TemporaryDirectory() as d:
|
|
clip = os.path.join(d, "clip_001.mp4")
|
|
upsert_clip_annotation(d, clip, "dog barking")
|
|
with open(os.path.join(d, "dataset.json")) as f:
|
|
entries = json.load(f)
|
|
assert len(entries) == 1
|
|
assert entries[0]["label"] == "dog barking"
|
|
assert entries[0]["path"] == clip
|
|
|
|
def test_upsert_appends_new_clips():
|
|
with tempfile.TemporaryDirectory() as d:
|
|
upsert_clip_annotation(d, os.path.join(d, "clip_001.mp4"), "dog barking")
|
|
upsert_clip_annotation(d, os.path.join(d, "clip_002.mp4"), "cat meowing")
|
|
with open(os.path.join(d, "dataset.json")) as f:
|
|
entries = json.load(f)
|
|
assert len(entries) == 2
|
|
|
|
def test_upsert_replaces_existing():
|
|
with tempfile.TemporaryDirectory() as d:
|
|
clip = os.path.join(d, "clip_001.mp4")
|
|
upsert_clip_annotation(d, clip, "dog barking")
|
|
upsert_clip_annotation(d, clip, "cat meowing")
|
|
with open(os.path.join(d, "dataset.json")) as f:
|
|
entries = json.load(f)
|
|
assert len(entries) == 1
|
|
assert entries[0]["label"] == "cat meowing"
|
|
|
|
def test_upsert_empty_label_skips():
|
|
with tempfile.TemporaryDirectory() as d:
|
|
upsert_clip_annotation(d, os.path.join(d, "clip_001.mp4"), "")
|
|
assert not os.path.exists(os.path.join(d, "dataset.json"))
|
|
|
|
def test_upsert_missing_folder_creates_it():
|
|
with tempfile.TemporaryDirectory() as d:
|
|
nested = os.path.join(d, "subdir", "deep")
|
|
upsert_clip_annotation(nested, os.path.join(nested, "clip_001.mp4"), "dog barking")
|
|
assert os.path.exists(os.path.join(nested, "dataset.json"))
|
|
|
|
def test_db_stores_label_and_category():
|
|
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
|
|
path = f.name
|
|
try:
|
|
db = ProcessedDB(path)
|
|
db.add("video.mp4", 0.0, "/out/clip_001.mp4", label="dog barking", category="Animal")
|
|
row = db._con.execute(
|
|
"SELECT label, category FROM processed WHERE filename = ?", ("video.mp4",)
|
|
).fetchone()
|
|
assert row == ("dog barking", "Animal")
|
|
finally:
|
|
os.unlink(path)
|
|
|
|
|
|
def test_db_get_group_returns_all_sub_clips():
|
|
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
|
|
path = f.name
|
|
try:
|
|
db = ProcessedDB(path)
|
|
db.add("video.mp4", 10.0, "/out/vid_001/clip_001_0.mp4")
|
|
db.add("video.mp4", 10.0, "/out/vid_001/clip_001_1.mp4")
|
|
db.add("video.mp4", 10.0, "/out/vid_001/clip_001_2.mp4")
|
|
group = db.get_group("/out/vid_001/clip_001_0.mp4")
|
|
assert len(group) == 3
|
|
assert "/out/vid_001/clip_001_0.mp4" in group
|
|
assert "/out/vid_001/clip_001_2.mp4" in group
|
|
finally:
|
|
os.unlink(path)
|
|
|
|
|
|
def test_db_get_group_isolates_by_start_time():
|
|
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
|
|
path = f.name
|
|
try:
|
|
db = ProcessedDB(path)
|
|
db.add("video.mp4", 10.0, "/out/vid_001/clip_001_0.mp4")
|
|
db.add("video.mp4", 10.0, "/out/vid_001/clip_001_1.mp4")
|
|
db.add("video.mp4", 30.0, "/out/vid_001/clip_002_0.mp4")
|
|
group = db.get_group("/out/vid_001/clip_001_0.mp4")
|
|
assert len(group) == 2
|
|
finally:
|
|
os.unlink(path)
|
|
|
|
|
|
def test_db_delete_group_removes_all():
|
|
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
|
|
path = f.name
|
|
try:
|
|
db = ProcessedDB(path)
|
|
db.add("video.mp4", 10.0, "/out/vid_001/clip_001_0.mp4")
|
|
db.add("video.mp4", 10.0, "/out/vid_001/clip_001_1.mp4")
|
|
db.add("video.mp4", 30.0, "/out/vid_001/clip_002_0.mp4")
|
|
deleted = db.delete_group("/out/vid_001/clip_001_0.mp4")
|
|
assert len(deleted) == 2
|
|
# clip_002 should still exist
|
|
markers = db.get_markers("video.mp4")
|
|
assert len(markers) == 1
|
|
assert markers[0][0] == 30.0
|
|
finally:
|
|
os.unlink(path)
|
|
|
|
|
|
def test_db_get_group_disabled():
|
|
db = ProcessedDB("/no/such/directory/8cut.db")
|
|
assert db.get_group("/out/clip_001.mp4") == []
|
|
|
|
|
|
def test_db_delete_group_disabled():
|
|
db = ProcessedDB("/no/such/directory/8cut.db")
|
|
assert db.delete_group("/out/clip_001.mp4") == []
|
|
|
|
|
|
# --- Profiles ---
|
|
|
|
def test_db_markers_isolated_by_profile():
|
|
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
|
|
path = f.name
|
|
try:
|
|
db = ProcessedDB(path)
|
|
db.add("video.mp4", 10.0, "/out/a_001.mp4", profile="landscape")
|
|
db.add("video.mp4", 20.0, "/out/b_001.mp4", profile="portrait")
|
|
land = db.get_markers("video.mp4", profile="landscape")
|
|
port = db.get_markers("video.mp4", profile="portrait")
|
|
assert len(land) == 1
|
|
assert land[0][0] == 10.0
|
|
assert len(port) == 1
|
|
assert port[0][0] == 20.0
|
|
finally:
|
|
os.unlink(path)
|
|
|
|
|
|
def test_db_get_profiles():
|
|
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
|
|
path = f.name
|
|
try:
|
|
db = ProcessedDB(path)
|
|
assert db.get_profiles() == []
|
|
db.add("a.mp4", 0.0, "/out/a.mp4", profile="beta")
|
|
db.add("b.mp4", 0.0, "/out/b.mp4", profile="alpha")
|
|
db.add("c.mp4", 0.0, "/out/c.mp4", profile="beta")
|
|
profiles = db.get_profiles()
|
|
assert profiles == ["alpha", "beta"]
|
|
finally:
|
|
os.unlink(path)
|
|
|
|
|
|
def test_db_get_profiles_disabled():
|
|
db = ProcessedDB("/no/such/directory/8cut.db")
|
|
assert db.get_profiles() == []
|
|
|
|
|
|
def test_db_default_profile_backward_compat():
|
|
"""Existing tests pass without explicit profile — defaults to 'default'."""
|
|
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
|
|
path = f.name
|
|
try:
|
|
db = ProcessedDB(path)
|
|
db.add("video.mp4", 5.0, "/out/clip.mp4")
|
|
markers = db.get_markers("video.mp4") # no profile arg
|
|
assert len(markers) == 1
|
|
assert markers[0][0] == 5.0
|
|
assert db.get_profiles() == ["default"]
|
|
finally:
|
|
os.unlink(path)
|
|
|
|
|
|
# --- resolve_keyframe ---
|
|
|
|
def test_resolve_keyframe_empty():
|
|
assert resolve_keyframe([], 5.0) is None
|
|
|
|
def test_resolve_keyframe_before_first():
|
|
kfs = [(3.0, 0.5, None, False, False)]
|
|
assert resolve_keyframe(kfs, 1.0) is None
|
|
|
|
def test_resolve_keyframe_exact():
|
|
kfs = [(2.0, 0.3, "9:16", True, False)]
|
|
assert resolve_keyframe(kfs, 2.0) == (2.0, 0.3, "9:16", True, False)
|
|
|
|
def test_resolve_keyframe_between():
|
|
kfs = [
|
|
(1.0, 0.2, None, False, False),
|
|
(5.0, 0.8, "1:1", False, True),
|
|
]
|
|
assert resolve_keyframe(kfs, 3.0) == (1.0, 0.2, None, False, False)
|
|
|
|
def test_resolve_keyframe_after_last():
|
|
kfs = [
|
|
(1.0, 0.2, None, False, False),
|
|
(5.0, 0.8, "1:1", False, True),
|
|
]
|
|
assert resolve_keyframe(kfs, 10.0) == (5.0, 0.8, "1:1", False, True)
|
|
|
|
def test_resolve_keyframe_tolerance():
|
|
kfs = [(4.0, 0.5, None, True, True)]
|
|
assert resolve_keyframe(kfs, 3.96) == (4.0, 0.5, None, True, True)
|
|
|
|
|
|
# --- apply_keyframes_to_jobs ---
|
|
|
|
def test_apply_keyframes_no_keyframes():
|
|
jobs = [(0.0, "/out/a", None, 0.5), (3.0, "/out/b", None, 0.5)]
|
|
result = apply_keyframes_to_jobs(jobs, [], base_center=0.5, base_ratio=None,
|
|
base_rand_p=True, base_rand_s=False)
|
|
assert result == [
|
|
(0.0, "/out/a", None, 0.5, True, False),
|
|
(3.0, "/out/b", None, 0.5, True, False),
|
|
]
|
|
|
|
def test_apply_keyframes_with_keyframes():
|
|
kfs = [
|
|
(0.0, 0.3, "9:16", True, False),
|
|
(4.0, 0.7, None, False, True),
|
|
]
|
|
jobs = [
|
|
(0.0, "/out/a", None, 0.5),
|
|
(3.0, "/out/b", None, 0.5),
|
|
(6.0, "/out/c", None, 0.5),
|
|
]
|
|
result = apply_keyframes_to_jobs(jobs, kfs, base_center=0.5, base_ratio=None,
|
|
base_rand_p=False, base_rand_s=False)
|
|
assert result == [
|
|
(0.0, "/out/a", "9:16", 0.3, True, False),
|
|
(3.0, "/out/b", "9:16", 0.3, True, False),
|
|
(6.0, "/out/c", None, 0.7, False, True),
|
|
]
|
|
|
|
def test_apply_keyframes_before_first_uses_base():
|
|
kfs = [(5.0, 0.8, "1:1", False, True)]
|
|
jobs = [(1.0, "/out/a", None, 0.5)]
|
|
result = apply_keyframes_to_jobs(jobs, kfs, base_center=0.5, base_ratio="4:5",
|
|
base_rand_p=True, base_rand_s=False)
|
|
assert result == [(1.0, "/out/a", "4:5", 0.5, True, False)]
|
|
|
|
|
|
# --- LTX-2 legal-frame math (core/ltx2.py) ---
|
|
|
|
from core.ltx2 import is_legal_frames, nearest_legal_frames, frames_for_duration, duration_for_frames, legal_frames
|
|
|
|
def test_ltx2_is_legal():
|
|
assert is_legal_frames(201) and is_legal_frames(9) and is_legal_frames(25)
|
|
assert not is_legal_frames(200) and not is_legal_frames(8)
|
|
|
|
def test_ltx2_nearest():
|
|
assert nearest_legal_frames(200) == 201 # 200 -> nearest 8k+1
|
|
assert nearest_legal_frames(196) == 193
|
|
assert nearest_legal_frames(5) == 9 # floor at 9
|
|
|
|
def test_ltx2_duration_roundtrip():
|
|
assert duration_for_frames(201, 25) == 201 / 25
|
|
assert frames_for_duration(8.0, 25) == 201 # 200 -> 201
|
|
|
|
def test_ltx2_legal_series():
|
|
s = legal_frames(min_f=9, max_f=33)
|
|
assert s == [9, 17, 25, 33]
|
|
|
|
|
|
# --- LTX-2 ffmpeg params (target_fps, snap32, frames) ---
|
|
|
|
def test_ffmpeg_ltx2_fps_and_frames():
|
|
cmd = build_ffmpeg_command("/in/v.mp4", 0.0, "/out/c.mp4",
|
|
short_side=512, target_fps=25, frames=201)
|
|
assert "-r" in cmd and cmd[cmd.index("-r")+1] == "25"
|
|
assert "-frames:v" in cmd and cmd[cmd.index("-frames:v")+1] == "201"
|
|
vf = cmd[cmd.index("-vf")+1]
|
|
assert "fps=25" in vf
|
|
|
|
def test_ffmpeg_ltx2_snap32_crop():
|
|
cmd = build_ffmpeg_command("/in/v.mp4", 0.0, "/out/c.mp4",
|
|
short_side=512, snap32=True)
|
|
vf = cmd[cmd.index("-vf")+1]
|
|
assert "crop=trunc(iw/32)*32:trunc(ih/32)*32" in vf
|
|
|
|
def test_ffmpeg_foley_unchanged():
|
|
cmd = build_ffmpeg_command("/in/v.mp4", 0.0, "/out/c.mp4", short_side=256)
|
|
assert "-r" not in cmd and "-frames:v" not in cmd
|
|
assert "crop=trunc" not in cmd[cmd.index("-vf")+1]
|
|
|
|
|
|
# --- LTX-2 audio extract frame-exact duration ---
|
|
|
|
def test_audio_extract_ltx2_duration():
|
|
frames, fps = 201, 25
|
|
cmd = build_audio_extract_command("/in/v.mp4", 0.0, "/out/clip_001",
|
|
duration=frames / fps)
|
|
assert "-t" in cmd
|
|
assert cmd[cmd.index("-t") + 1] == str(frames / fps)
|
|
|
|
|
|
# --- waveform peak reduction + region decode ---
|
|
|
|
def test_peaks_downsamples_to_bucket_count():
|
|
from core.waveform import peaks
|
|
import numpy as np
|
|
samples = np.sin(np.linspace(0, 100, 10000)).astype("float32")
|
|
p = peaks(samples, buckets=64)
|
|
assert len(p) == 64
|
|
assert all(0.0 <= v <= 1.0 for v in p)
|
|
import pytest
|
|
assert max(p) == pytest.approx(1.0)
|
|
|
|
|
|
def test_peaks_empty_returns_zeros():
|
|
from core.waveform import peaks
|
|
assert peaks(None, buckets=16) == [0.0] * 16
|
|
import numpy as np
|
|
assert peaks(np.zeros(0, dtype="float32"), buckets=8) == [0.0] * 8
|
|
|
|
|
|
def test_load_region_samples_bad_path_returns_empty():
|
|
from core.waveform import load_region_samples
|
|
import numpy as np
|
|
out = load_region_samples("/no/such/file.mp4", 0.0, 1.0)
|
|
assert isinstance(out, np.ndarray)
|
|
assert out.size == 0
|
|
|
|
|
|
def test_time_pixel_roundtrip():
|
|
from core.waveform import t_to_x, x_to_t
|
|
assert x_to_t(0, 400, 10.0, 4.0) == 10.0
|
|
assert x_to_t(400, 400, 10.0, 4.0) == 14.0
|
|
assert x_to_t(200, 400, 10.0, 4.0) == 12.0
|
|
assert t_to_x(12.0, 400, 10.0, 4.0) == 200
|
|
for x in (0, 37, 200, 399):
|
|
assert abs(t_to_x(x_to_t(x, 400, 10.0, 4.0), 400, 10.0, 4.0) - x) <= 1
|
|
|
|
|
|
def test_time_pixel_guards():
|
|
from core.waveform import t_to_x, x_to_t
|
|
assert x_to_t(50, 0, 10.0, 4.0) == 10.0 # zero width -> view_start
|
|
assert x_to_t(50, 400, 10.0, 0.0) == 10.0 # zero span -> view_start
|
|
assert t_to_x(12.0, 400, 10.0, 0.0) == 0 # zero span -> 0
|