diff --git a/lib/variation_checker.py b/lib/variation_checker.py index 2604a941..47829f10 100644 --- a/lib/variation_checker.py +++ b/lib/variation_checker.py @@ -56,13 +56,20 @@ def check_scene_variation(scenes: list[dict[str, Any]]) -> dict[str, Any]: suggestions.append("Mix wide establishing shots with close-ups for visual rhythm.") # --- Check 2: Consecutive same-size shots --- - consecutive_same = 0 + # Track the longest actual run of identical shot sizes. Summing every equal + # adjacent pair across the whole plan would count non-consecutive groups + # (e.g. wide,wide,cu,cu,med,med -> 3 pairs) as a single "3 consecutive" run. + longest_run = 1 if shot_sizes else 0 + current_run = 1 for i in range(1, len(shot_sizes)): if shot_sizes[i] == shot_sizes[i-1] and shot_sizes[i] != "unspecified": - consecutive_same += 1 - if consecutive_same >= 3: + current_run += 1 + longest_run = max(longest_run, current_run) + else: + current_run = 1 + if longest_run >= 3: violations.append( - f"{consecutive_same} consecutive same-size shots. " + f"{longest_run} consecutive same-size shots. " f"Vary shot sizes between scenes for editorial rhythm." ) diff --git a/schemas/artifacts/source_media_review.schema.json b/schemas/artifacts/source_media_review.schema.json index 7e603502..0e89cdc8 100644 --- a/schemas/artifacts/source_media_review.schema.json +++ b/schemas/artifacts/source_media_review.schema.json @@ -9,6 +9,7 @@ "version": { "type": "string", "const": "1.0" }, "files": { "type": "array", + "description": "Reviewed source files. Empty when no user media was supplied (or none could be reviewed) — a valid 'fully generated production' state that review_source_media reports explicitly.", "items": { "type": "object", "required": ["path", "media_type", "reviewed"], @@ -64,7 +65,7 @@ }, "additionalProperties": false }, - "minItems": 1 + "minItems": 0 }, "summary": { "type": "string", diff --git a/tests/lib/test_source_media_review_empty.py b/tests/lib/test_source_media_review_empty.py new file mode 100644 index 00000000..54ed39fa --- /dev/null +++ b/tests/lib/test_source_media_review_empty.py @@ -0,0 +1,29 @@ +"""Regression test for source_media_review empty-files artifact validity. + +review_source_media deliberately returns an artifact with files:[] when no user +media was supplied (or none could be reviewed) — a valid "fully generated +production" state. The schema declared files.minItems: 1, so that intended +artifact failed its own schema validation. +""" + +import sys +from pathlib import Path + +PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent +sys.path.insert(0, str(PROJECT_ROOT)) + +from lib.source_media_review import review_source_media # noqa: E402 +from schemas.artifacts import validate_artifact # noqa: E402 + + +def test_no_source_media_produces_schema_valid_artifact(tmp_path): + art = review_source_media([tmp_path / "does-not-exist.mp4"], {}) + assert art["files"] == [] + # Must not raise — this is a legitimate no-source-media artifact. + validate_artifact("source_media_review", art) + + +def test_no_files_at_all_is_schema_valid(): + art = review_source_media([], {}) + assert art["files"] == [] + validate_artifact("source_media_review", art) diff --git a/tests/lib/test_variation_checker_runs.py b/tests/lib/test_variation_checker_runs.py new file mode 100644 index 00000000..fcec201f --- /dev/null +++ b/tests/lib/test_variation_checker_runs.py @@ -0,0 +1,37 @@ +"""Regression test for check_scene_variation consecutive-run counting. + +The "consecutive same-size shots" check summed every equal adjacent pair across +the whole plan instead of measuring the longest actual run, so an editorially +varied plan of separate 2-shot groups (wide,wide,cu,cu,med,med) falsely tripped +a "3 consecutive same-size shots" violation. +""" + +import sys +from pathlib import Path + +PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent +sys.path.insert(0, str(PROJECT_ROOT)) + +from lib.variation_checker import check_scene_variation # noqa: E402 + + +def _scenes(sizes): + return [{"shot_language": {"shot_size": s}} for s in sizes] + + +def test_non_consecutive_same_size_pairs_do_not_trip_run_check(): + # Three separate 2-shot groups — longest run is 2, not 3. + res = check_scene_variation(_scenes(["wide", "wide", "cu", "cu", "medium", "medium"])) + assert not any("consecutive same-size" in v for v in res["violations"]) + + +def test_true_run_of_three_is_flagged(): + res = check_scene_variation(_scenes(["wide", "wide", "wide", "cu", "medium"])) + assert any("3 consecutive same-size" in v for v in res["violations"]) + + +def test_unspecified_shots_do_not_form_a_run(): + res = check_scene_variation( + _scenes(["unspecified", "unspecified", "unspecified", "unspecified"]) + ) + assert not any("consecutive same-size" in v for v in res["violations"]) diff --git a/tests/tools/test_audio_mixer_ducking.py b/tests/tools/test_audio_mixer_ducking.py new file mode 100644 index 00000000..ec060e81 --- /dev/null +++ b/tests/tools/test_audio_mixer_ducking.py @@ -0,0 +1,89 @@ +"""Regression tests for audio_mixer full_mix ducking filtergraph. + +The ducking branch built an `acopy[speech_dup]` filter whose output pad was +never consumed, leaving the FFmpeg filtergraph with a dangling output. FFmpeg +rejects that, so `full_mix` with the most common shape — a single narration +track plus one music bed, with ducking enabled (the default) — always failed. +""" + +import shutil +import subprocess +import sys +from pathlib import Path + +import pytest + +PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent +sys.path.insert(0, str(PROJECT_ROOT)) + +from tools.audio.audio_mixer import AudioMixer # noqa: E402 + +pytestmark = pytest.mark.skipif( + shutil.which("ffmpeg") is None, reason="ffmpeg required for full_mix" +) + + +def _sine(path: Path, freq: int, dur: int) -> None: + subprocess.run( + ["ffmpeg", "-y", "-f", "lavfi", "-i", f"sine=frequency={freq}:duration={dur}", str(path)], + capture_output=True, + check=True, + timeout=30, + ) + + +def _has_audio(path: Path) -> bool: + out = subprocess.run( + ["ffprobe", "-v", "error", "-select_streams", "a", + "-show_entries", "stream=codec_type", "-of", "csv=p=0", str(path)], + capture_output=True, text=True, timeout=30, + ) + return "audio" in out.stdout + + +def test_full_mix_single_narration_plus_music_with_ducking(tmp_path): + speech = tmp_path / "speech.wav" + music = tmp_path / "music.wav" + _sine(speech, 440, 2) + _sine(music, 220, 3) + out = tmp_path / "mixed.wav" + + result = AudioMixer().execute( + { + "operation": "full_mix", + "tracks": [ + {"path": str(speech), "role": "speech"}, + {"path": str(music), "role": "music"}, + ], + "ducking": {"enabled": True}, + "output_path": str(out), + } + ) + + assert result.success is True, result.error + assert out.exists() and _has_audio(out) + + +def test_full_mix_multi_narration_plus_music_with_ducking(tmp_path): + s1, s2 = tmp_path / "s1.wav", tmp_path / "s2.wav" + music = tmp_path / "music.wav" + _sine(s1, 440, 2) + _sine(s2, 330, 2) + _sine(music, 220, 3) + out = tmp_path / "mixed_multi.wav" + + result = AudioMixer().execute( + { + "operation": "full_mix", + "tracks": [ + {"path": str(s1), "role": "speech"}, + {"path": str(s2), "role": "speech"}, + {"path": str(music), "role": "music"}, + ], + "ducking": {"enabled": True}, + "output_path": str(out), + } + ) + + assert result.success is True, result.error + assert out.exists() and _has_audio(out) diff --git a/tools/audio/audio_mixer.py b/tools/audio/audio_mixer.py index dba08b0b..04a052e3 100644 --- a/tools/audio/audio_mixer.py +++ b/tools/audio/audio_mixer.py @@ -492,17 +492,22 @@ class AudioMixer(BaseTool): duck_enabled = ducking.get("enabled", True) if isinstance(ducking, dict) else bool(ducking) if duck_enabled and speech_tracks and music_tracks: - # Mix speech tracks together first + # Build ONE speech stream, then split it into two independent + # branches: one feeds the sidechain compressor as the ducking key, + # the other is mixed into the final output. A filtergraph label may + # only be consumed once, so reusing the same speech label for both + # the sidechain key and the output mix is invalid on stricter ffmpeg + # builds (e.g. the Linux ffmpeg on CI). asplit makes the fork explicit. speech_indices = list(range(len(speech_tracks))) speech_labels = "".join(f"[a{i}]" for i in speech_indices) if len(speech_tracks) > 1: filter_parts.append( - f"{speech_labels}amix=inputs={len(speech_tracks)}:duration=longest[speech_mix]" + f"{speech_labels}amix=inputs={len(speech_tracks)}:duration=longest[speech_all]" ) - speech_out = "[speech_mix]" else: - speech_out = f"[a{speech_indices[0]}]" + filter_parts.append(f"[a{speech_indices[0]}]acopy[speech_all]") + filter_parts.append("[speech_all]asplit=2[speech_key][speech_out]") # Mix music tracks together music_start = len(speech_tracks) @@ -517,42 +522,20 @@ class AudioMixer(BaseTool): else: music_in = f"[a{music_indices[0]}]" - # Apply sidechain ducking + # Apply sidechain ducking — music is compressed, [speech_key] is the key duck_params = ducking if isinstance(ducking, dict) else {} attack = duck_params.get("attack_ms", 200) / 1000 release = duck_params.get("release_ms", 500) / 1000 music_vol = duck_params.get("music_volume_during_speech", 0.15) filter_parts.append( - f"{music_in}{speech_out}sidechaincompress=" + f"{music_in}[speech_key]sidechaincompress=" f"threshold=0.02:ratio=9:attack={attack}:release={release}:" f"level_sc=1:mix=0.9[ducked_music];" f"[ducked_music]volume={music_vol * 3}[music_out]" ) - # Duplicate speech for final mix (sidechain consumes it as key) - filter_parts.append( - f"{speech_out}acopy[speech_dup]" if speech_out.startswith("[a") else "" - ) - # Re-mix speech path: we need speech audio in the output too - # Simpler approach: use amix on original speech and ducked music - # Reset: use a cleaner approach — amerge the speech mix and ducked music - # Actually, let's rebuild. The sidechain approach above uses speech as - # the key signal but doesn't consume it from the output chain. - # FFmpeg sidechaincompress: input 0 = audio to compress, input 1 = key signal - # So music is compressed, speech signal is the key. We need to mix them. - # Remove the last filter_part (the acopy that may be empty) - if filter_parts and filter_parts[-1] == "": - filter_parts.pop() - - # Build speech mix for output separately - if len(speech_tracks) > 1: - # speech_mix already exists, make a copy for output - filter_parts.append(f"{speech_labels}amix=inputs={len(speech_tracks)}:duration=longest[speech_out]") - else: - filter_parts.append(f"[a{speech_indices[0]}]acopy[speech_out]") - - # Final mix: speech_out + music_out + # Final mix: the other speech branch + ducked music mix_label = "[speech_out][music_out]amix=inputs=2:duration=longest[premix]" # Add SFX if present