castle-batch-v1/tests/test_batch_prepare.py

import json
import subprocess
from pathlib import Path
from batch_pipeline import prepare_media
from test_batch_tasks import CloudFiles
from castle_pipeline.batch_tasks import TaskPlanner


def test_preparation_persists_model_native_frames_audio_and_relative_times(tmp_path):
    source=tmp_path/'source.mp4'
    subprocess.run(['ffmpeg','-v','error','-f','lavfi','-i','color=c=green:s=320x180:r=50',
                    '-f','lavfi','-i','sine=frequency=440:sample_rate=48000','-t','1.2','-c:v','libx264',
                    '-threads','1','-c:a','aac',str(source)],check=True)
    cloud=CloudFiles();planner=TaskPlanner(cloud,tmp_path/'scratch',tmp_path/'output')
    cfg={'gcs_prefix':'gs://b/run','fps':1.,'clip_seconds':30.,'max_dim':1440,'start_clip':0,'max_clips':2,'review':True}
    rows=prepare_media(source,{'source_id':'fixture','viewpoint':'egocentric'},cfg,planner)
    assert len(rows)==1
    meta=json.loads(cloud.files[rows[0]['metadata_uri']])
    assert meta['duration_sec']==1.2 and meta['frame_times_sec']==[0.,1.]
    assert len(meta['native_frames'])==len(meta['frames'])==2
    assert all(uri in cloud.files for uri in meta['frames']+meta['native_frames']+[meta['audio_uri']])
    assert not list((tmp_path/'scratch').rglob('*.jpg'))