diff --git a/actions_lib/ffmpeg.py b/actions_lib/ffmpeg.py index 8980d50..51e8d59 100644 --- a/actions_lib/ffmpeg.py +++ b/actions_lib/ffmpeg.py @@ -76,14 +76,59 @@ }) -def _require_finite_number(value: Any, name: str) -> float: +def _require_finite_number( + value: Any, name: str, *, min_value: float | None = None +) -> float: if ( isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value) ): raise ValueError(f'{name} must be a finite number: {value!r}') - return float(value) + val = float(value) + if min_value is not None and val < min_value: + if min_value == 0.0: + raise ValueError(f'{name} must be non-negative: {value!r}') + raise ValueError(f'{name} must be >= {min_value}: {value!r}') + return val + + +def _clean_duration( + duration: float, + available_duration: float, + source_duration: float, + target_fps: float, +) -> float: + """Clamps clip duration to known footage and aligns it to frame boundaries. + + Args: + duration: Caller-requested duration in seconds (<= 0 means use remaining). + available_duration: Remaining footage after skip_time in seconds. + source_duration: Probed source container duration in seconds (0 if unknown). + target_fps: Output frame rate used to quantize duration to whole frames. + + Returns: + Frame-aligned clip duration in seconds. + + Raises: + ValueError: If duration <= 0 and the source duration is unknown. + """ + frame_duration = 1.0 / target_fps + if duration > 0: + effective_duration = ( + min(duration, available_duration) + if source_duration > 0 + else duration + ) + elif source_duration > 0: + effective_duration = min(source_duration, available_duration) + else: + raise ValueError( + 'Explicit positive duration is required when video source duration is' + ' unknown' + ) + total_frames = max(1, round(effective_duration / frame_duration)) + return total_frames * frame_duration def _require_int(value: Any, name: str) -> int: @@ -225,7 +270,7 @@ def add_video( Returns: the instance of FFMPEG so that calls can be chained. """ - skip_time = _require_finite_number(skip_time, 'skip_time') + skip_time = _require_finite_number(skip_time, 'skip_time', min_value=0.0) duration = _require_finite_number(duration, 'duration') if transition is not None: if transition not in _XFADE_TRANSITIONS: @@ -233,22 +278,40 @@ def add_video( transition_overlap = _require_finite_number( 0.0 if transition_overlap is None else transition_overlap, 'transition_overlap', + min_value=0.0, ) else: if transition_overlap is not None: transition_overlap = _require_finite_number( - transition_overlap, 'transition_overlap' + transition_overlap, 'transition_overlap', min_value=0.0 ) properties = get_video_properties(path) + if properties['duration'] > 0 and skip_time >= properties['duration']: + raise ValueError( + f'skip_time ({skip_time}) must be less than video duration' + f" ({properties['duration']})" + ) + available_duration = max(0.0, properties['duration'] - skip_time) + + fps_changed = False if properties['fps'] > self.target_fps: self.target_fps = properties['fps'] - # Recompute the duration to be an integer multiple of frames - frame_duration = 1.0 / self.target_fps - if duration > 0: - total_frames = round(duration / frame_duration) - clean_duration = total_frames * frame_duration - else: - clean_duration = properties['duration'] + fps_changed = True + + clean_duration = _clean_duration( + duration, available_duration, properties['duration'], self.target_fps + ) + + if fps_changed: + for item in self.inputs: + if item['type'] == 'video': + item['duration'] = _clean_duration( + item['raw_duration'], + item['available_duration'], + item['source_duration'], + self.target_fps, + ) + self.inputs.append({ 'type': 'video', 'path': path, @@ -257,6 +320,9 @@ def add_video( 'has_audio': properties['has_audio'] and include_audio, 'transition': transition, 'transition_overlap': transition_overlap, + 'raw_duration': duration, + 'available_duration': available_duration, + 'source_duration': properties['duration'], }) return self @@ -274,8 +340,10 @@ def add_audio( Returns: the instance of FFMPEG so that calls can be chained. """ - start_time = _require_finite_number(start_time, 'start_time') - skip_time = _require_finite_number(skip_time, 'skip_time') + start_time = _require_finite_number( + start_time, 'start_time', min_value=0.0 + ) + skip_time = _require_finite_number(skip_time, 'skip_time', min_value=0.0) duration = _require_finite_number(duration, 'duration') self.inputs.append({ 'type': 'audio', @@ -314,8 +382,14 @@ def add_image( TODO: use the start_time to start the clip at the right place. """ - start_time = _require_finite_number(start_time, 'start_time') + start_time = _require_finite_number( + start_time, 'start_time', min_value=0.0 + ) duration = _require_finite_number(duration, 'duration') + if duration <= 0: + raise ValueError( + f'Image overlay requires an explicit positive duration: {duration!r}' + ) offset_x = _require_int(offset_x, 'offset_x') offset_y = _require_int(offset_y, 'offset_y') width = _require_int(width, 'width') diff --git a/actions_lib/test/test_ffmpeg.py b/actions_lib/test/test_ffmpeg.py index 90d7613..e881d61 100644 --- a/actions_lib/test/test_ffmpeg.py +++ b/actions_lib/test/test_ffmpeg.py @@ -15,6 +15,9 @@ """Tests for FFMPEG utility class.""" import json +from pathlib import Path +import subprocess +import tempfile import unittest from unittest import mock @@ -250,5 +253,466 @@ def test_convert_video_rejects_invalid_resolution_or_extension( self.ffmpeg.convert_video('input.mp4', bad_ext) + @mock.patch('actions_lib.ffmpeg.get_video_properties') + def test_add_video_duration_clamp_and_controls(self, mock_get_props): + mock_get_props.return_value = { + 'duration': 10.0, + 'dimensions': '1280:720', + 'fps': 30.0, + 'has_audio': True, + } + + # Case 1: skip_time=3, duration=-1 on a 10 s source yields 7.0 s (not 10.0) + ffmpeg1 = FFMPEG() + ffmpeg1.add_video( + path='clip.mp4', + skip_time=3.0, + duration=-1.0, + transition=None, + transition_overlap=0, + ) + self.assertEqual(ffmpeg1.inputs[0]['duration'], 7.0) + + # Case 2: skip_time=3, duration=10 also yields 7.0 s (second branch) + ffmpeg2 = FFMPEG() + ffmpeg2.add_video( + path='clip.mp4', + skip_time=3.0, + duration=10.0, + transition=None, + transition_overlap=0, + ) + self.assertEqual(ffmpeg2.inputs[0]['duration'], 7.0) + + # Control 1: skip_time=0, duration=-1 is unchanged (10.0 s) + ffmpeg3 = FFMPEG() + ffmpeg3.add_video( + path='clip.mp4', + skip_time=0.0, + duration=-1.0, + transition=None, + transition_overlap=0, + ) + self.assertEqual(ffmpeg3.inputs[0]['duration'], 10.0) + + # Control 2: skip_time=3, duration=4 (< available 7 s) is unchanged (4.0 s) + ffmpeg4 = FFMPEG() + ffmpeg4.add_video( + path='clip.mp4', + skip_time=3.0, + duration=4.0, + transition=None, + transition_overlap=0, + ) + self.assertEqual(ffmpeg4.inputs[0]['duration'], 4.0) + + @mock.patch('actions_lib.ffmpeg.get_video_properties') + def test_unknown_source_duration_uses_requested_duration( + self, mock_get_props + ): + mock_get_props.return_value = { + 'duration': 0.0, + 'dimensions': '1280:720', + 'fps': 30.0, + 'has_audio': True, + } + + ffmpeg = FFMPEG() + ffmpeg.add_video( + path='unknown-duration.mp4', + skip_time=3.0, + duration=4.0, + transition=None, + transition_overlap=0, + ) + + self.assertEqual(ffmpeg.inputs[0]['skip'], 3.0) + self.assertEqual(ffmpeg.inputs[0]['duration'], 4.0) + + @mock.patch('actions_lib.ffmpeg.get_video_properties') + def test_unknown_source_duration_requires_explicit_duration( + self, mock_get_props + ): + mock_get_props.return_value = { + 'duration': 0.0, + 'dimensions': '1280:720', + 'fps': 30.0, + 'has_audio': True, + } + + ffmpeg = FFMPEG() + with self.assertRaisesRegex(ValueError, 'Explicit positive duration'): + ffmpeg.add_video( + path='unknown-duration.mp4', + skip_time=0.0, + duration=-1.0, + transition=None, + transition_overlap=0, + ) + self.assertEqual(ffmpeg.inputs, []) + + @mock.patch('actions_lib.ffmpeg.get_video_properties') + def test_negative_times_and_excessive_skip_raise(self, mock_get_props): + mock_get_props.return_value = { + 'duration': 10.0, + 'dimensions': '1280:720', + 'fps': 30.0, + 'has_audio': True, + } + + # Negative skip_time in add_video + with self.assertRaises(ValueError): + self.ffmpeg.add_video( + path='clip.mp4', + skip_time=-1.0, + duration=5.0, + transition=None, + transition_overlap=0, + ) + + # Negative transition_overlap in add_video + with self.assertRaises(ValueError): + self.ffmpeg.add_video( + path='clip.mp4', + skip_time=0.0, + duration=5.0, + transition='fade', + transition_overlap=-0.5, + ) + + # skip_time >= source duration in add_video + with self.assertRaises(ValueError): + self.ffmpeg.add_video( + path='clip.mp4', + skip_time=10.0, + duration=-1.0, + transition=None, + transition_overlap=0, + ) + with self.assertRaises(ValueError): + self.ffmpeg.add_video( + path='clip.mp4', + skip_time=12.0, + duration=5.0, + transition=None, + transition_overlap=0, + ) + + # Negative start_time and skip_time in add_audio + with self.assertRaises(ValueError): + self.ffmpeg.add_audio( + path='audio.mp3', start_time=-1.0, skip_time=0.0, duration=5.0 + ) + with self.assertRaises(ValueError): + self.ffmpeg.add_audio( + path='audio.mp3', start_time=0.0, skip_time=-1.0, duration=5.0 + ) + + # Negative start_time and duration in add_image + with self.assertRaises(ValueError): + self.ffmpeg.add_image( + path='img.png', + start_time=-0.5, + duration=1.0, + offset_x=0, + offset_y=0, + width=100, + height=100, + ) + with self.assertRaises(ValueError): + self.ffmpeg.add_image( + path='img.png', + start_time=0.0, + duration=-1.0, + offset_x=0, + offset_y=0, + width=100, + height=100, + ) + + @mock.patch('actions_lib.ffmpeg.get_video_properties') + def test_target_fps_order_independent_determinism(self, mock_get_props): + props = { + 'clip24.mp4': { + 'duration': 5.0, + 'dimensions': '160:90', + 'fps': 24.0, + 'has_audio': False, + }, + 'clip60.mp4': { + 'duration': 5.0, + 'dimensions': '160:90', + 'fps': 60.0, + 'has_audio': False, + }, + } + mock_get_props.side_effect = lambda path: dict(props[path]) + + f1 = FFMPEG() + f1.add_video( + path='clip24.mp4', + skip_time=0, + duration=1.05, + transition=None, + transition_overlap=0, + ) + f1.add_video( + path='clip60.mp4', + skip_time=0, + duration=1.05, + transition=None, + transition_overlap=0, + ) + + f2 = FFMPEG() + f2.add_video( + path='clip60.mp4', + skip_time=0, + duration=1.05, + transition=None, + transition_overlap=0, + ) + f2.add_video( + path='clip24.mp4', + skip_time=0, + duration=1.05, + transition=None, + transition_overlap=0, + ) + + self.assertEqual(f1.target_fps, 60.0) + self.assertEqual(f2.target_fps, 60.0) + self.assertEqual(f1.inputs[0]['duration'], f2.inputs[1]['duration']) + self.assertEqual(f1.inputs[1]['duration'], f2.inputs[0]['duration']) + + def test_real_ffmpeg_render_av_sync(self): + with tempfile.TemporaryDirectory() as tmp_dir: + clip1 = Path(tmp_dir) / 'clip1.mp4' + clip2 = Path(tmp_dir) / 'clip2.mp4' + for p, c, f in [(clip1, 'red', 440), (clip2, 'blue', 880)]: + subprocess.run( + [ + 'ffmpeg', + '-y', + '-v', + 'error', + '-f', + 'lavfi', + '-i', + f'color=c={c}:s=160x90:r=24', + '-f', + 'lavfi', + '-i', + f'sine=frequency={f}:duration=3', + '-t', + '3', + '-c:v', + 'libx264', + '-pix_fmt', + 'yuv420p', + '-c:a', + 'aac', + '-shortest', + str(p), + ], + check=True, + ) + + # 1. Single clip with skip_time=1.0, duration=-1.0 + ffmpeg = FFMPEG().set_resolution('160:90') + ffmpeg.add_video( + path=str(clip1), + skip_time=1.0, + duration=-1.0, + transition=None, + transition_overlap=0, + ) + out_single = Path(tmp_dir) / 'out_single.mp4' + ffmpeg.combine( + str(out_single), shortest_stream=False, encoding_speed=8, video_crf=28 + ) + + probe_cmd = [ + 'ffprobe', + '-v', + 'error', + '-show_entries', + 'stream=codec_type,duration:format=duration', + '-of', + 'json', + str(out_single), + ] + res = subprocess.run( + probe_cmd, capture_output=True, text=True, check=True + ) + probe_data = json.loads(res.stdout) + + v_dur, a_dur = None, None + for s in probe_data.get('streams', []): + if s.get('codec_type') == 'video': + v_dur = float(s['duration']) + elif s.get('codec_type') == 'audio': + a_dur = float(s['duration']) + + self.assertIsNotNone(v_dur) + self.assertIsNotNone(a_dur) + self.assertAlmostEqual(v_dur, 2.0, places=2) + self.assertAlmostEqual(a_dur, 2.0, places=2) + self.assertAlmostEqual(abs(a_dur - v_dur), 0.0, places=2) + + # 2. Concat with crossfade transition and skip_time + ffmpeg_xfade = FFMPEG().set_resolution('160:90') + ffmpeg_xfade.add_video( + path=str(clip1), + skip_time=1.0, + duration=-1.0, + transition=None, + transition_overlap=0, + ) + ffmpeg_xfade.add_video( + path=str(clip2), + skip_time=1.0, + duration=-1.0, + transition='fade', + transition_overlap=0.5, + ) + out_xfade = Path(tmp_dir) / 'out_xfade.mp4' + ffmpeg_xfade.combine( + str(out_xfade), shortest_stream=False, encoding_speed=8, video_crf=28 + ) + + probe_cmd[-1] = str(out_xfade) + res = subprocess.run( + probe_cmd, capture_output=True, text=True, check=True + ) + probe_data = json.loads(res.stdout) + + v_dur, a_dur = None, None + for s in probe_data.get('streams', []): + if s.get('codec_type') == 'video': + v_dur = float(s['duration']) + elif s.get('codec_type') == 'audio': + a_dur = float(s['duration']) + + self.assertIsNotNone(v_dur) + self.assertIsNotNone(a_dur) + self.assertAlmostEqual(v_dur, 3.5, places=2) + self.assertAlmostEqual(a_dur, 3.5, places=2) + self.assertAlmostEqual(abs(a_dur - v_dur), 0.0, places=2) + + def test_add_image_requires_explicit_positive_duration(self): + for bad_duration in (0, 0.0, -1, -5.5): + with self.assertRaisesRegex( + ValueError, 'Image overlay requires an explicit positive duration' + ): + self.ffmpeg.add_image( + path='img.png', + start_time=0.0, + duration=bad_duration, + offset_x=0, + offset_y=0, + width=100, + height=100, + ) + + ffmpeg = FFMPEG() + ffmpeg.add_image( + path='img.png', + start_time=1.0, + duration=4.0, + offset_x=10, + offset_y=10, + width=100, + height=100, + ) + self.assertEqual(len(ffmpeg.inputs), 1) + self.assertEqual(ffmpeg.inputs[0]['start'], 1.0) + self.assertEqual(ffmpeg.inputs[0]['end'], 5.0) + + @mock.patch('actions_lib.ffmpeg.get_media_duration') + def test_add_audio_permits_negative_duration_fallback(self, mock_dur): + mock_dur.return_value = 15.25 + ffmpeg = FFMPEG() + ffmpeg.add_audio('audio.mp3', start_time=0.0, skip_time=0.0, duration=-1) + self.assertEqual(ffmpeg.inputs[0]['duration'], 15.25) + mock_dur.assert_called_once_with('audio.mp3') + + @mock.patch('actions_lib.ffmpeg.get_video_properties') + @mock.patch('actions_lib.ffmpeg.get_media_duration') + def test_shipped_workflow_examples_accepted_by_ffmpeg( + self, mock_dur, mock_props + ): + mock_props.return_value = { + 'duration': 30.0, + 'dimensions': '1280:720', + 'fps': 30.0, + 'has_audio': True, + } + mock_dur.return_value = 30.0 + + repo_root = Path(__file__).resolve().parent.parent.parent + examples_dir = repo_root / 'workflow_examples' / 'input' + self.assertTrue( + examples_dir.is_dir(), + f'Workflow examples directory not found: {examples_dir}', + ) + + example_files = sorted(examples_dir.glob('*_arrangement.json')) + self.assertGreaterEqual( + len(example_files), + 2, + f'Expected at least 2 shipped arrangement files in {examples_dir}', + ) + + for jf in example_files: + with open(jf) as f: + arrangement = json.load(f) + + ffmpeg = FFMPEG() + for arr in arrangement: + skip_time = arr.get('skip_time', 0) + duration = arr.get('duration', -1) + offset_x = arr.get('offset_x', 0) + offset_y = arr.get('offset_y', 0) + local_path = '/mock/' + arr['file_path'] + + if arr['file_type'] == 'video': + transition = arr.get('transition') + transition_overlap = arr.get('transition_overlap') + include_audio = arr.get('include_audio', True) + if transition and transition_overlap is None: + transition_overlap = 0.5 + ffmpeg.add_video( + path=local_path, + skip_time=skip_time, + duration=duration, + transition=transition, + transition_overlap=transition_overlap, + include_audio=include_audio, + ) + elif arr['file_type'] == 'audio': + ffmpeg.add_audio( + local_path, arr['start_time'], skip_time, duration + ) + elif arr['file_type'] == 'image': + ffmpeg.add_image( + path=local_path, + start_time=arr['start_time'], + duration=duration, + offset_x=offset_x, + offset_y=offset_y, + width=arr['width'], + height=arr.get('height', -1), + ) + else: + self.fail(f'Unexpected file_type in {jf}: {arr.get("file_type")}') + + self.assertEqual( + len(ffmpeg.inputs), + len(arrangement), + f'All entries in {jf.name} should be accepted', + ) + + if __name__ == '__main__': unittest.main()