158 lines
6.5 KiB
Python
158 lines
6.5 KiB
Python
"""Verifies the stream-ends offset estimate: it lands on the known offset of
|
|
each synthetic fixture, it feeds Preview/Export exactly like a
|
|
marker-derived offset does (including with no markers set at all), and
|
|
Use Marker Offset brings the marker-derived value back afterwards without
|
|
re-marking -- while the manual marker workflow itself behaves as before.
|
|
|
|
Also covers the real-world shape this feature exists for, which none of
|
|
the fixtures in test_media have: both tracks start at t=0 but the audio
|
|
ends early (it started capturing late, and the recorder stopped both
|
|
together). Generated on the fly as a 60s clip, long enough that the probe
|
|
only reads the tail window instead of the whole file.
|
|
"""
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
from pathlib import Path
|
|
|
|
from PyQt6.QtWidgets import QApplication
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
import main as app_main
|
|
from stream_edges import probe_stream_edges, TAIL_WINDOW_SEC
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
MEDIA = ROOT / "test_media"
|
|
|
|
|
|
def wait_for(predicate, qapp, timeout=20):
|
|
deadline = time.time() + timeout
|
|
while not predicate() and time.time() < deadline:
|
|
qapp.processEvents()
|
|
time.sleep(0.05)
|
|
return predicate()
|
|
|
|
|
|
def load(win, qapp, path):
|
|
win._load(str(path))
|
|
qapp.processEvents()
|
|
time.sleep(0.2)
|
|
qapp.processEvents()
|
|
|
|
|
|
def estimate(win, qapp):
|
|
win.estimate_button.click()
|
|
qapp.processEvents()
|
|
assert wait_for(lambda: win.estimate_button.isEnabled(), qapp), "estimate did not finish in time"
|
|
|
|
|
|
def make_audio_ends_early_clip(path: Path):
|
|
subprocess.run([
|
|
"ffmpeg", "-y", "-v", "error",
|
|
"-f", "lavfi", "-i", "testsrc=size=320x240:rate=30:duration=60",
|
|
"-f", "lavfi", "-i", "sine=frequency=1000:duration=59",
|
|
"-c:v", "libx264", "-preset", "ultrafast", "-c:a", "aac",
|
|
str(path),
|
|
], check=True)
|
|
|
|
|
|
def main():
|
|
# -- Probe alone, against every fixture's known offset.
|
|
expected = {
|
|
"sync_test_INSYNC.mp4": 0,
|
|
"sync_test_OUTOFSYNC_audio+350ms.mp4": -350,
|
|
"sync_test_OUTOFSYNC_video+350ms.mp4": 350,
|
|
}
|
|
for name, want in expected.items():
|
|
edges = probe_stream_edges(str(MEDIA / name))
|
|
print(f"{name}: end-based {edges.end_delay_ms:+.1f} ms, start-based {edges.start_delay_ms:+.1f} ms")
|
|
assert abs(edges.end_delay_ms - want) < 5, f"{name}: expected ~{want} ms"
|
|
print("PASS: end-based estimate matches every fixture's known offset")
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
clip = Path(tmp) / "audio_ends_early.mp4"
|
|
make_audio_ends_early_clip(clip)
|
|
assert 60 > TAIL_WINDOW_SEC, "clip should be long enough to exercise the tail-window read"
|
|
edges = probe_stream_edges(str(clip))
|
|
print(f"audio-ends-early clip: {edges}, end-based {edges.end_delay_ms:+.1f} ms")
|
|
assert abs(edges.end_delay_ms - 1000) < 30, "expected ~+1000 ms (delay audio)"
|
|
assert edges.estimates_disagree, "starts both at 0 but ends 1s apart -- should flag disagreement"
|
|
print("PASS: audio-ends-early clip estimates ~+1000 ms and flags the start/end disagreement")
|
|
|
|
qapp = QApplication(sys.argv)
|
|
win = app_main.MainWindow()
|
|
win.show()
|
|
qapp.processEvents()
|
|
|
|
assert not win.estimate_button.isEnabled(), "estimate needs a loaded file"
|
|
assert not win.use_marker_offset_button.isEnabled()
|
|
|
|
# -- Estimate with no markers: enables Preview/Export on its own.
|
|
load(win, qapp, MEDIA / "sync_test_OUTOFSYNC_video+350ms.mp4")
|
|
assert win.estimate_button.isEnabled() and win.estimate_action.isEnabled()
|
|
assert not win.export_button.isEnabled() and not win.preview_button.isEnabled()
|
|
estimate(win, qapp)
|
|
print(f"offset label after estimate: {win.offset_label.text()!r}")
|
|
assert win.offset.delay_ms == 350
|
|
assert win.override_spinbox.value() == 350 and win.override_spinbox.isEnabled()
|
|
assert win.export_button.isEnabled() and win.preview_button.isEnabled()
|
|
assert "estimated from stream ends" in win.offset_label.text()
|
|
assert not win.use_marker_offset_button.isEnabled(), "no markers yet, nothing to go back to"
|
|
assert win.video_mark is None and win.audio_mark is None, "estimate must not touch the markers"
|
|
print("PASS: estimate with no markers sets the offset and enables Preview/Export")
|
|
|
|
# -- Preview works from an estimate with no markers (centered on the playhead).
|
|
win.source_panel.player.seek(5.0)
|
|
qapp.processEvents()
|
|
win.preview_button.click()
|
|
qapp.processEvents()
|
|
assert wait_for(lambda: win.video_tabs.currentIndex() == 1, qapp), "preview did not build"
|
|
print("PASS: Preview builds from an estimated offset with no markers set")
|
|
win.video_tabs.setCurrentIndex(0)
|
|
|
|
# -- Manual markers behave as before and override the estimate.
|
|
load(win, qapp, MEDIA / "sync_test_OUTOFSYNC_audio+350ms.mp4")
|
|
assert win.offset_label.text() == "Offset: — (mark both events)"
|
|
win.source_panel.player.seek(5.000)
|
|
qapp.processEvents()
|
|
win._mark_video()
|
|
win.source_panel.player.seek(5.350)
|
|
qapp.processEvents()
|
|
win._mark_audio()
|
|
marker_label = win.offset_label.text()
|
|
print(f"offset label from markers: {marker_label!r}")
|
|
assert abs(win.offset.delay_ms - (-350)) < 40
|
|
assert marker_label == win.offset.describe()
|
|
assert win.use_marker_offset_button.isEnabled()
|
|
marker_delay = win.offset.delay_ms
|
|
print("PASS: manual markers set the offset exactly as before")
|
|
|
|
# -- Estimate replaces it; Use Marker Offset restores it without re-marking.
|
|
estimate(win, qapp)
|
|
assert win.offset.delay_ms == -349, f"got {win.offset.delay_ms}"
|
|
assert win.override_spinbox.value() == -349
|
|
assert win.video_mark is not None and win.audio_mark is not None, "estimate must not clear markers"
|
|
win.use_marker_offset_button.click()
|
|
qapp.processEvents()
|
|
assert win.offset.delay_ms == marker_delay
|
|
assert win.override_spinbox.value() == round(marker_delay)
|
|
assert win.offset_label.text() == marker_label
|
|
print("PASS: Use Marker Offset restores the marker-derived offset after an estimate")
|
|
|
|
# -- Loading a new file resets everything, including Use Marker Offset.
|
|
load(win, qapp, MEDIA / "sync_test_INSYNC.mp4")
|
|
assert not win.use_marker_offset_button.isEnabled()
|
|
assert not win.export_button.isEnabled()
|
|
estimate(win, qapp)
|
|
assert win.offset.delay_ms == 0
|
|
print("PASS: new file resets marker state; in-sync file estimates 0 ms")
|
|
|
|
win.close()
|
|
qapp.processEvents()
|
|
print("\nSTREAM ENDS ESTIMATE SMOKE TEST PASSED")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|