diff --git a/ffmpeg_export.py b/ffmpeg_export.py new file mode 100644 index 0000000..5a29e1d --- /dev/null +++ b/ffmpeg_export.py @@ -0,0 +1,86 @@ +"""Lossless A/V shift export. Ports the ffmpeg logic already proven out in +the `ffdelay` bash script (see PROJECT_BRIEF.md) -- reuse it rather than +reinvent it. +""" +import subprocess +from dataclasses import dataclass +from typing import Optional + + +class FfmpegExportError(RuntimeError): + """ffmpeg exited non-zero. Carries stderr so the caller can surface the + real reason (e.g. a codec/container that can't stream-copy) instead of + silently falling back to a re-encode -- that fallback is an explicit + non-goal for v1. + """ + + def __init__(self, message: str, stderr: str): + super().__init__(message) + self.stderr = stderr + + +@dataclass(frozen=True) +class ExportResult: + command: list + stderr: str + + +def export_synced( + input_path: str, + delay_ms: float, + output_path: str, + clip_start: Optional[str] = None, + clip_duration: Optional[float] = None, +) -> ExportResult: + """Stream-copy `input_path` to `output_path` with the audio and video + tracks re-aligned by `delay_ms` (see sync_logic.py for the sign + convention). `clip_start`/`clip_duration` extract a short segment + instead of processing the whole file, for fast iteration -- mirrors + ffdelay's optional clip mode. + + ffmpeg's `-itsoffset` doesn't accept a negative value the way this needs, + so a negative delay_ms is applied by putting the offset on the *other* + input (video) instead of trying to negate it -- these are genuinely two + different command shapes, not one parametrized by sign. + """ + abs_delay_sec = abs(delay_ms) / 1000.0 + # Python's float formatting always includes the leading zero (unlike + # `bc`, which emits ".500" for sub-1 values and made ffmpeg reject the + # duration outright) -- no special-casing needed here. + offset_str = f"{abs_delay_sec:.3f}" + + clip_args = [] + if clip_start is not None and clip_duration is not None: + # Applied to BOTH inputs so the extracted clip covers the same + # real-world time window in each, regardless of which one also + # carries -itsoffset. + clip_args = ["-ss", clip_start, "-t", str(clip_duration)] + + if delay_ms < 0: + video_input_args = ["-itsoffset", offset_str, *clip_args, "-i", input_path] + audio_input_args = [*clip_args, "-i", input_path] + else: + video_input_args = [*clip_args, "-i", input_path] + audio_input_args = ["-itsoffset", offset_str, *clip_args, "-i", input_path] + + command = [ + "ffmpeg", "-y", + *video_input_args, + *audio_input_args, + "-map", "0:v", "-map", "1:a", + "-c:v", "copy", "-c:a", "copy", + output_path, + ] + + proc = subprocess.run(command, capture_output=True, text=True) + if proc.returncode != 0: + raise FfmpegExportError( + f"ffmpeg exited with code {proc.returncode}", proc.stderr + ) + + # stderr is returned even on success: ffmpeg logs things like + # "sps_id out of range" / "Invalid NAL unit size" / "missing picture in + # access unit" on stream-copy operations from screen recordings/live + # captures. These are harmless source-file quirks, not failures -- the + # caller should log them, not treat them as errors. + return ExportResult(command=command, stderr=proc.stderr) diff --git a/scripts/test_export_roundtrip.py b/scripts/test_export_roundtrip.py new file mode 100644 index 0000000..82ee130 --- /dev/null +++ b/scripts/test_export_roundtrip.py @@ -0,0 +1,91 @@ +"""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() diff --git a/sync_logic.py b/sync_logic.py new file mode 100644 index 0000000..2ff4e3e --- /dev/null +++ b/sync_logic.py @@ -0,0 +1,45 @@ +"""Pure marker/offset math -- no UI, no ffmpeg, no mpv. Easily testable +in isolation per the project's suggested build order. +""" +from dataclasses import dataclass + + +@dataclass(frozen=True) +class SyncOffset: + delay_ms: float + + @property + def target(self) -> str: + """Which track the export step needs to delay to fix sync.""" + if self.delay_ms > 0: + return "audio" + if self.delay_ms < 0: + return "video" + return "none" + + def describe(self) -> str: + if self.delay_ms == 0: + return "Offset: 0 ms (already in sync)" + sign = "+" if self.delay_ms > 0 else "" + action = "audio will be delayed" if self.delay_ms > 0 else "video will be delayed" + return f"Offset: {sign}{self.delay_ms:.0f} ms ({action})" + + +def compute_offset(video_mark_sec: float, audio_mark_sec: float) -> SyncOffset: + """video_mark_sec / audio_mark_sec are timeline positions (seconds) in + the *currently loaded* file where the same real-world event (clap/flash/ + beep) was observed on each track. + + delay_ms = (video_mark - audio_mark) * 1000 + + Sign convention (matches PROJECT_BRIEF.md and ffmpeg_export.py): + - delay_ms > 0: the audio event appears *earlier* in the timeline than + the video event, i.e. audio is ahead -- it needs to be delayed + (shifted later) to line up with video. + - delay_ms < 0: the audio event appears *later* than the video event, + i.e. audio is already behind -- delaying it further would be wrong. + Instead the video gets delayed by the same amount, which catches + video up to where the (late) audio already is. + - delay_ms == 0: already in sync. + """ + return SyncOffset(delay_ms=(video_mark_sec - audio_mark_sec) * 1000.0) diff --git a/test_media/README.md b/test_media/README.md index b1c3caf..94f4ae2 100644 --- a/test_media/README.md +++ b/test_media/README.md @@ -10,14 +10,24 @@ validating marker capture, offset math, and lossless export. - `sync_test_OUTOFSYNC_audio+350ms.mp4` — same content, but the audio track has been shifted +350ms later relative to video (via `-itsoffset 0.350` on the audio input, stream-copied). The flash event stays at video t=5.000s; the beep event - now appears at t=5.350s. + now appears at t=5.350s. Exercises the negative-delay_ms / "delay video" branch + in ffmpeg_export.py. + +- `sync_test_OUTOFSYNC_video+350ms.mp4` — same content, but the *video* track has + been shifted +350ms later instead (via `-itsoffset 0.350` on the video input). + The beep stays at audio t=5.000s; the flash now appears at t=5.350s. Exercises + the positive-delay_ms / "delay audio" branch in ffmpeg_export.py. **Ground truth for the app's marker/offset workflow:** - Mark Video at the flash → ~5.000s - Mark Audio at the beep → ~5.350s -- Expected computed `delay_ms` = +350 (audio needs to be delayed to match video) -- Exporting `sync_test_OUTOFSYNC_audio+350ms.mp4` with delay_ms = +350 should - reproduce (bit-for-bit, modulo container overhead) `sync_test_INSYNC.mp4`'s sync. +- Expected computed `delay_ms` = (video_mark - audio_mark) * 1000 = **-350** + (audio is already the *late* one here, so the fix is to delay the video by + 350ms to catch up to it -- not to delay the audio further. Negative + delay_ms → offset the video input, per sync_logic.py's sign convention.) +- Exporting `sync_test_OUTOFSYNC_audio+350ms.mp4` with delay_ms = -350 should + land both events at t=5.350s in the output (not back at 5.000s -- the + absolute timeline shifts, only the *relative* sync matters). Frame at exactly t=5.000s is frame index 150 (30fps) — useful for checking frame-step precision lands exactly on it. diff --git a/test_media/sync_test_OUTOFSYNC_video+350ms.mp4 b/test_media/sync_test_OUTOFSYNC_video+350ms.mp4 new file mode 100644 index 0000000..1f9b437 Binary files /dev/null and b/test_media/sync_test_OUTOFSYNC_video+350ms.mp4 differ diff --git a/tests/test_sync_logic.py b/tests/test_sync_logic.py new file mode 100644 index 0000000..c70fccf --- /dev/null +++ b/tests/test_sync_logic.py @@ -0,0 +1,38 @@ +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) +from sync_logic import compute_offset + + +class TestComputeOffset(unittest.TestCase): + def test_audio_ahead_needs_delay(self): + # Flash seen at 5.000s, beep heard at 4.500s -> audio is running + # ahead of video, so audio needs to be delayed to catch up. + offset = compute_offset(video_mark_sec=5.000, audio_mark_sec=4.500) + self.assertAlmostEqual(offset.delay_ms, 500.0) + self.assertEqual(offset.target, "audio") + + def test_audio_behind_needs_video_delay(self): + # This is our synthetic test fixture: audio was shifted +350ms + # later than video, so the fix is to delay video to catch up to + # the already-late audio -- NOT to delay audio further. + offset = compute_offset(video_mark_sec=5.000, audio_mark_sec=5.350) + self.assertAlmostEqual(offset.delay_ms, -350.0) + self.assertEqual(offset.target, "video") + + def test_already_in_sync(self): + offset = compute_offset(video_mark_sec=5.000, audio_mark_sec=5.000) + self.assertEqual(offset.delay_ms, 0.0) + self.assertEqual(offset.target, "none") + + def test_describe_formatting(self): + self.assertIn("audio will be delayed", compute_offset(5.5, 5.0).describe()) + self.assertIn("video will be delayed", compute_offset(5.0, 5.5).describe()) + self.assertIn("+500", compute_offset(5.5, 5.0).describe()) + self.assertIn("-500", compute_offset(5.0, 5.5).describe()) + + +if __name__ == "__main__": + unittest.main()