67 lines
3.7 KiB
Python
Executable File
67 lines
3.7 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Validate a persisted clip before any source cleanup (ffmpeg/ffprobe required)."""
|
|
import json
|
|
import pathlib
|
|
import subprocess
|
|
import sys
|
|
|
|
saved_path = pathlib.Path(sys.argv[1])
|
|
saved = json.loads(saved_path.read_text())
|
|
clip = saved_path.with_suffix('.mp4')
|
|
session_path = saved_path.parent.parent / 'sessions' / str(saved['source_session']) / 'session.json'
|
|
session = json.loads(session_path.read_text())
|
|
seconds = lambda d: d['secs'] + d['nanos'] / 1e9
|
|
start, end = seconds(saved['start']), seconds(saved['end'])
|
|
assert end > start
|
|
pieces = [s for s in session['segments'] if s['available'] and seconds(s['start']) < end and seconds(s['end']) > start]
|
|
assert len(pieces) >= 2, 'Export must cross a segment boundary'
|
|
|
|
def probe(path):
|
|
return json.loads(subprocess.check_output(['ffprobe', '-v', 'error', '-show_streams', '-show_packets', '-of', 'json', str(path)]))
|
|
|
|
source = probe(session_path.parent / pieces[0]['path'])
|
|
output = probe(clip)
|
|
video = next(s for s in output['streams'] if s['codec_type'] == 'video')
|
|
source_video = next(s for s in source['streams'] if s['codec_type'] == 'video')
|
|
assert video['codec_name'] == source_video['codec_name']
|
|
pts = sorted(float(p['pts_time']) for p in output['packets'] if p['stream_index'] == video['index'])
|
|
fps = pieces[0]['frames_per_second']
|
|
for a, b in zip(pts, pts[1:]):
|
|
assert abs(b-a-1/fps) < 2e-6, ('Video gap/duplicate', a, b, fps)
|
|
assert abs(pts[0]) < 1/fps
|
|
assert abs(pts[-1]+1/fps-(end-start)) < 1/fps+0.002, (pts[-1], start, end)
|
|
# Metadata records independent AAC epochs, even if earlier files are cleaned up later.
|
|
runs={}
|
|
for segment in session['segments']:
|
|
if not segment['complete'] or not segment['available']:
|
|
continue
|
|
run=str(pathlib.Path(segment['path']).parent)
|
|
facts=probe(session_path.parent/segment['path'])
|
|
audios=[s for s in facts['streams'] if s['codec_type']=='audio']
|
|
totals=runs.setdefault(run,[0]*len(audios))
|
|
assert segment['audio_start_samples']==totals,(segment['path'],segment['audio_start_samples'],totals)
|
|
for i,stream in enumerate(audios):
|
|
totals[i]+=1024*len([p for p in facts['packets'] if p['stream_index']==stream['index']])
|
|
|
|
for ordinal,stream in enumerate([s for s in output['streams'] if s['codec_type']=='audio']):
|
|
if stream['codec_type'] == 'audio':
|
|
timestamps = [float(p['pts_time']) for p in output['packets'] if p['stream_index'] == stream['index']]
|
|
step = 1024 / int(stream['sample_rate'])
|
|
assert timestamps
|
|
import math
|
|
source_ordinal=ordinal if saved['separate_audio'] or len([s for s in source['streams'] if s['codec_type']=='audio'])!=3 else 2
|
|
first=pieces[0]
|
|
run_base=seconds(first['start'])-first['frame_start']/fps
|
|
phase=run_base+first['audio_start_samples'][source_ordinal]/int(stream['sample_rate'])-start
|
|
expected_first=phase+max(0,math.ceil(-phase/step))*step
|
|
assert abs(timestamps[0]-expected_first)<1/int(stream['sample_rate'])+2e-6,('A/V phase',timestamps[0],expected_first)
|
|
for a, b in zip(timestamps, timestamps[1:]):
|
|
assert abs(b-a-step) < 2e-6, ('Audio gap/duplicate', a, b)
|
|
source_audio = [s for s in source['streams'] if s['codec_type'] == 'audio']
|
|
expected = 2 if saved['separate_audio'] and len(source_audio) == 3 else 1
|
|
assert len([s for s in output['streams'] if s['codec_type'] == 'audio']) == expected
|
|
result = subprocess.run(['ffmpeg', '-v', 'error', '-xerror', '-i', str(clip), '-f', 'null', '-'], capture_output=True)
|
|
assert result.returncode == 0 and not result.stderr, result.stderr.decode()
|
|
assert saved['clip']['size_bytes'] == clip.stat().st_size
|
|
print(f'PASS clip {clip.name}: {video["codec_name"]}, {len(pts)} CFR frames, continuous audio, zero decode errors, {end-start:.3f}s')
|