ThePendulum
f634dda5bb
sync_logic.py: pure delay_ms computation with the documented sign convention. Caught and fixed a sign error in my own test-fixture ground-truth doc while writing the unit tests: the audio+350ms fixture actually requires delay_ms=-350 (delay video), not +350, since the audio is already the late one. ffmpeg_export.py: ports ffdelay's two-branch command structure (offset video input vs audio input depending on delay_ms sign) plus its documented quirks (leading-zero float formatting, -ss/-t applied to both inputs for clip mode, non-fatal stream-copy warnings surfaced via stderr rather than swallowed). Added an end-to-end integration test (scripts/test_export_roundtrip.py) that independently re-detects event timing in the exported output via ffmpeg scene-change/silence detection (not the same code path used to build the export) and confirms both the positive- and negative-delay branches bring drift down from ~350ms to under 1ms.
92 lines
3.5 KiB
Python
92 lines
3.5 KiB
Python
"""Integration test for ffmpeg_export.py against the synthetic ground-truth
|
|
fixture (build-order step 3, and a down payment on step 5).
|
|
|
|
Runs the real marker->offset->export pipeline against
|
|
test_media/sync_test_OUTOFSYNC_audio+350ms.mp4, then independently
|
|
re-detects where the flash and beep landed in the *output* file using
|
|
ffmpeg scene-change / silence detection (not the same code path used to
|
|
build the export, so this isn't circular) and checks they now line up.
|
|
"""
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
from sync_logic import compute_offset
|
|
from ffmpeg_export import export_synced
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
OUT_DIR = ROOT / "test_media" / "seek_validation"
|
|
|
|
|
|
def find_video_event_time(path: Path) -> float:
|
|
"""Scene-change detection: the flash causes two sharp scene changes
|
|
(onset and offset), so the first hit is the flash's start time."""
|
|
proc = subprocess.run(
|
|
[
|
|
"ffmpeg", "-i", str(path),
|
|
"-vf", "select='gt(scene,0.15)',showinfo",
|
|
"-f", "null", "-",
|
|
],
|
|
capture_output=True, text=True,
|
|
)
|
|
times = [float(m) for m in re.findall(r"pts_time:([\d.]+)", proc.stderr)]
|
|
if not times:
|
|
raise RuntimeError(f"no scene change detected in {path}:\n{proc.stderr}")
|
|
return times[0]
|
|
|
|
|
|
def find_audio_event_time(path: Path) -> float:
|
|
"""silencedetect: the near-silent background audio going loud for the
|
|
beep produces a silence_end at (approximately) the beep's onset."""
|
|
proc = subprocess.run(
|
|
[
|
|
"ffmpeg", "-i", str(path),
|
|
"-af", "silencedetect=noise=-20dB:d=0.1",
|
|
"-f", "null", "-",
|
|
],
|
|
capture_output=True, text=True,
|
|
)
|
|
times = [float(m) for m in re.findall(r"silence_end: ([\d.]+)", proc.stderr)]
|
|
if not times:
|
|
raise RuntimeError(f"no silence_end detected in {path}:\n{proc.stderr}")
|
|
return times[0]
|
|
|
|
|
|
def run_case(fixture_name: str, expected_target: str, output_name: str):
|
|
input_path = ROOT / "test_media" / fixture_name
|
|
output_path = OUT_DIR / output_name
|
|
|
|
print(f"Input fixture: {fixture_name}")
|
|
video_mark = find_video_event_time(input_path)
|
|
audio_mark = find_audio_event_time(input_path)
|
|
print(f" detected flash at t={video_mark:.3f}s, beep at t={audio_mark:.3f}s")
|
|
|
|
offset = compute_offset(video_mark_sec=video_mark, audio_mark_sec=audio_mark)
|
|
print(f" {offset.describe()}")
|
|
assert offset.target == expected_target, f"expected {expected_target} branch, got {offset.target}"
|
|
|
|
result = export_synced(str(input_path), offset.delay_ms, str(output_path))
|
|
print(f" ran: {' '.join(result.command)}")
|
|
|
|
after_video_t = find_video_event_time(output_path)
|
|
after_audio_t = find_audio_event_time(output_path)
|
|
print(f" post-export: flash at t={after_video_t:.3f}s, beep at t={after_audio_t:.3f}s")
|
|
|
|
drift_ms = abs(after_video_t - after_audio_t) * 1000
|
|
print(f" drift after export: {drift_ms:.1f}ms")
|
|
assert drift_ms < 34, f"export did not fix sync: {drift_ms:.1f}ms drift remains (expected < 1 frame ~33ms)"
|
|
print(f" PASS ({expected_target}-delay branch)\n")
|
|
|
|
|
|
def main():
|
|
OUT_DIR.mkdir(exist_ok=True)
|
|
run_case("sync_test_OUTOFSYNC_audio+350ms.mp4", "video", "_export_roundtrip_negative.mp4")
|
|
run_case("sync_test_OUTOFSYNC_video+350ms.mp4", "audio", "_export_roundtrip_positive.mp4")
|
|
print("PASS: both ffmpeg command branches (positive and negative delay) round-trip correctly.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|