#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ ============================================================================= KIEU STORY — EMPIRICAL ADVERSARIAL TEST SUITE: M3 AUDIO CONTINUITY ENGINE ============================================================================= Agent: teamwork_preview_challenger (M3 Challenger 1) Target: 05_Production_Pipeline/audio_continuity_engine.py Empirical Stress-Testing Criteria: 1. Two-Pass EBU R128 loudness normalization produces output compliant with: - Integrated Loudness: -14.0 LUFS ± 0.5 LUFS - True Peak: <= -1.0 dBTP 2. Silent audio (-inf) does not crash FFmpeg in measure or normalization passes. 3. Video without audio stream (-an) concatenated with standard audio video: - Does not crash FFmpeg - Produces 48kHz stereo AAC stream 4. Mixed channel counts (mono + stereo) preserve stereo layout without downmixing. 5. Micro-fade Mode A (boundary_smoothing) blends boundaries with zero timeline shrinkage. 6. Extreme sample rates & active gain staging headroom protection. ============================================================================= """ import os import sys import json import math import shutil import tempfile import subprocess import unittest from pathlib import Path from typing import Dict, Any, List, Optional, Tuple # Ensure UTF-8 on Windows if sys.platform == "win32": try: sys.stdout.reconfigure(encoding="utf-8") sys.stderr.reconfigure(encoding="utf-8") except Exception: pass BASE_DIR = Path(__file__).resolve().parent.parent PIPELINE_DIR = BASE_DIR / "05_Production_Pipeline" if str(PIPELINE_DIR) not in sys.path: sys.path.insert(0, str(PIPELINE_DIR)) from audio_continuity_engine import AudioContinuityEngine, get_ffmpeg, get_ffprobe class TestAdversarialM3AudioEngine(unittest.TestCase): """ Adversarial empirical stress tests for AudioContinuityEngine. """ @classmethod def setUpClass(cls): cls.ffmpeg = get_ffmpeg() cls.ffprobe = get_ffprobe() cls.engine = AudioContinuityEngine() cls.telemetry_results = {} cls.results_json_path = BASE_DIR / "tests" / "adversarial_m3_audio_engine_results.json" @classmethod def tearDownClass(cls): # Save empirical telemetry try: with open(cls.results_json_path, "w", encoding="utf-8") as f: json.dump(cls.telemetry_results, f, indent=2, ensure_ascii=False) print(f"\n[✓] Saved empirical telemetry to: {cls.results_json_path}") except Exception as e: print(f"[!] Warning: failed to save telemetry: {e}") # ------------------------------------------------------------------------- # Helper methods for media generation and inspection # ------------------------------------------------------------------------- def _create_test_video( self, output_path: str, duration: float = 3.0, color: str = "black", audio_filter: Optional[str] = None, sample_rate: int = 48000, channels: int = 2, has_audio: bool = True ) -> str: """Helper to create deterministic test video with exact audio specs.""" cmd = [ self.ffmpeg, "-y", "-f", "lavfi", "-i", f"color=c={color}:s=320x240:d={duration}" ] if has_audio: if audio_filter: cmd.extend(["-filter_complex", audio_filter, "-map", "0:v", "-map", "[a_out]"]) else: cmd.extend([ "-f", "lavfi", "-i", f"sine=frequency=440:sample_rate={sample_rate}:duration={duration}", "-map", "0:v", "-map", "1:a" ]) cmd.extend([ "-c:v", "libx264", "-pix_fmt", "yuv420p", "-c:a", "aac", "-b:a", "192k", "-ar", str(sample_rate), "-ac", str(channels) ]) else: cmd.extend(["-an", "-c:v", "libx264", "-pix_fmt", "yuv420p"]) cmd.extend(["-t", str(duration), output_path]) res = subprocess.run(cmd, capture_output=True, text=True) if res.returncode != 0: raise RuntimeError(f"FFmpeg creation failed: {res.stderr}") return output_path def _probe_media(self, media_path: str) -> Dict[str, Any]: """Probes audio and video stream metadata via ffprobe.""" cmd = [ self.ffprobe, "-v", "error", "-show_entries", "stream=codec_type,codec_name,sample_rate,channels,channel_layout,duration:format=duration", "-of", "json", media_path ] res = subprocess.run(cmd, capture_output=True, text=True) self.assertEqual(res.returncode, 0, f"ffprobe failed: {res.stderr}") data = json.loads(res.stdout) streams = data.get("streams", []) fmt = data.get("format", {}) video_stream = next((s for s in streams if s.get("codec_type") == "video"), None) audio_stream = next((s for s in streams if s.get("codec_type") == "audio"), None) return { "format_duration": float(fmt.get("duration", 0.0)), "video_duration": float(video_stream.get("duration", fmt.get("duration", 0.0))) if video_stream else 0.0, "has_audio": audio_stream is not None, "audio_codec": audio_stream.get("codec_name") if audio_stream else None, "sample_rate": int(audio_stream.get("sample_rate", 0)) if audio_stream else 0, "channels": int(audio_stream.get("channels", 0)) if audio_stream else 0, "channel_layout": audio_stream.get("channel_layout", "") if audio_stream else "", "audio_duration": float(audio_stream.get("duration", fmt.get("duration", 0.0))) if audio_stream else 0.0 } def _measure_integrated_lufs_tp(self, media_path: str) -> Tuple[float, float, float]: """Measures Integrated Loudness (LUFS), True Peak (dBTP), and LRA (LU) via FFmpeg loudnorm JSON.""" cmd = [ self.ffmpeg, "-hide_banner", "-y", "-i", media_path, "-af", "loudnorm=I=-14:TP=-1.0:LRA=9:print_format=json", "-f", "null", "-" ] res = subprocess.run(cmd, capture_output=True, text=True) self.assertEqual(res.returncode, 0, f"Loudnorm pass failed: {res.stderr}") import re matches = re.findall(r'\{[\s\S]*?\}', res.stderr) self.assertTrue(len(matches) > 0, "Failed to parse loudnorm JSON from FFmpeg output") data = json.loads(matches[-1]) i_lufs = float(data.get("input_i", -99.0)) tp_dbtp = float(data.get("input_tp", -99.0)) lra_lu = float(data.get("input_lra", 0.0)) return i_lufs, tp_dbtp, lra_lu # ------------------------------------------------------------------------- # TEST 1: Two-Pass EBU R128 Loudness Normalization Compliance # ------------------------------------------------------------------------- def test_adv_01_two_pass_loudness_normalization_compliance(self): """ Adversarial Test 1: Verify Two-Pass EBU R128 loudness normalization produces: - Integrated Loudness: -14.0 LUFS ± 0.5 LUFS - True Peak: <= -1.0 dBTP across quiet, loud, complex dynamic, and 4-stem mixed signals. """ with tempfile.TemporaryDirectory() as td: cases = [ { "name": "quiet_speech_sim", "filter": "sine=f=300:r=48000:d=6,volume=-32dB[a_out]", "expected_pre_lufs": -45.0 }, { "name": "loud_music_sim", "filter": "sine=f=440:r=48000:d=6,volume=-2dB[a_out]", "expected_pre_lufs": -18.0 }, { "name": "complex_harmonic_ambient", "filter": "anoisesrc=d=8:c=pink:r=48000:a=0.15[n];sine=f=220:r=48000:d=8[s1];sine=f=660:r=48000:d=8[s2];[s1][s2]amix=inputs=2[s];[s][n]amix=inputs=2,volume=-14dB[a_out]", "expected_pre_lufs": -24.0 } ] telemetry_cases = [] for case in cases: in_path = os.path.join(td, f"in_{case['name']}.mp4") out_path = os.path.join(td, f"out_{case['name']}.mp4") # Generate video self._create_test_video(in_path, duration=6.0, audio_filter=case["filter"]) # Measure input pre_i, pre_tp, pre_lra = self._measure_integrated_lufs_tp(in_path) # Normalize via Two-Pass Linear ok = self.engine.normalize_loudness( in_path, out_path, target_lufs=-14.0, target_tp=-1.0, target_lra=9.0, two_pass=True ) self.assertTrue(ok, f"normalize_loudness failed for {case['name']}") self.assertTrue(os.path.exists(out_path)) # Measure output post_i, post_tp, post_lra = self._measure_integrated_lufs_tp(out_path) # Assertions: EBU R128 target: -14.0 ± 0.5 LUFS, True Peak <= -1.0 dBTP self.assertAlmostEqual( post_i, -14.0, delta=0.5, msg=f"Integrated Loudness {post_i:.2f} LUFS out of spec (-14.0 ± 0.5) for {case['name']}" ) self.assertLessEqual( post_tp, -0.99, msg=f"True Peak {post_tp:.2f} dBTP exceeds ceiling of -1.0 dBTP for {case['name']}" ) telemetry_cases.append({ "case": case["name"], "pre_lufs": pre_i, "pre_tp": pre_tp, "post_lufs": post_i, "post_tp": post_tp, "post_lra": post_lra, "compliant_lufs": abs(post_i - (-14.0)) <= 0.5, "compliant_tp": post_tp <= -0.99 }) # Sub-test: 4-Stem mixing normalization compliance v_base = os.path.join(td, "stem_video.mp4") bgm_wav = os.path.join(td, "stem_bgm.wav") amb_wav = os.path.join(td, "stem_amb.wav") diag_wav = os.path.join(td, "stem_diag.wav") out_master = os.path.join(td, "master_4stems.mp4") self._create_test_video(v_base, duration=6.0) subprocess.run([self.ffmpeg, "-y", "-f", "lavfi", "-i", "sine=f=180:r=48000:d=4", "-c:a", "pcm_s16le", bgm_wav], capture_output=True, check=True) subprocess.run([self.ffmpeg, "-y", "-f", "lavfi", "-i", "anoisesrc=d=4:c=pink:r=48000:a=0.1", "-c:a", "pcm_s16le", amb_wav], capture_output=True, check=True) subprocess.run([self.ffmpeg, "-y", "-f", "lavfi", "-i", "sine=f=440:r=48000:d=3", "-c:a", "pcm_s16le", diag_wav], capture_output=True, check=True) mix_ok = self.engine.mix_four_stems( video_path=v_base, output_path=out_master, stem1_bgm=bgm_wav, stem2_ambience=amb_wav, stem4_dialogue=diag_wav, target_lufs=-14.0, target_tp=-1.0, two_pass=True ) self.assertTrue(mix_ok, "mix_four_stems failed") self.assertTrue(os.path.exists(out_master)) stem_i, stem_tp, stem_lra = self._measure_integrated_lufs_tp(out_master) self.assertAlmostEqual(stem_i, -14.0, delta=0.5, msg=f"4-Stem mix LUFS {stem_i} not within -14.0 ± 0.5") self.assertLessEqual(stem_tp, -0.99, msg=f"4-Stem mix TP {stem_tp} exceeds -1.0 dBTP") telemetry_cases.append({ "case": "4_stem_mix_master", "post_lufs": stem_i, "post_tp": stem_tp, "post_lra": stem_lra, "compliant_lufs": abs(stem_i - (-14.0)) <= 0.5, "compliant_tp": stem_tp <= -0.99 }) self.__class__.telemetry_results["test_01_two_pass_compliance"] = telemetry_cases # ------------------------------------------------------------------------- # TEST 2: Silent Audio (-inf) Resilience # ------------------------------------------------------------------------- def test_adv_02_silent_audio_inf_resilience(self): """ Adversarial Test 2: Verify that purely silent audio (-inf) does NOT crash FFmpeg, parses correctly, and successfully executes normalization fallback. """ with tempfile.TemporaryDirectory() as td: silent_mp4 = os.path.join(td, "silent_input.mp4") norm_silent_mp4 = os.path.join(td, "silent_normalized.mp4") # Create pure silence MP4 subprocess.run([ self.ffmpeg, "-y", "-f", "lavfi", "-i", "color=c=black:s=320x240:d=4", "-f", "lavfi", "-i", "aevalsrc=0:d=4:s=48000:c=stereo", "-c:v", "libx264", "-c:a", "aac", "-ar", "48000", "-ac", "2", silent_mp4 ], capture_output=True, check=True) # 1. measure_loudness on pure silence meas = self.engine.measure_loudness(silent_mp4) self.assertIsNotNone(meas, "measure_loudness returned None for silent audio") self.assertTrue( meas.get("is_silent") is True, f"Expected is_silent == True, got {meas.get('is_silent')}" ) # Verify -inf values were safely recognized self.assertIn("-inf", [meas.get("input_i"), meas.get("input_tp"), meas.get("input_thresh")]) # 2. normalize_loudness on pure silence ok = self.engine.normalize_loudness( silent_mp4, norm_silent_mp4, target_lufs=-14.0, target_tp=-1.0, two_pass=True ) self.assertTrue(ok, "normalize_loudness must succeed on silent audio via fallback") self.assertTrue(os.path.exists(norm_silent_mp4), "Normalized silent video must exist") probe = self._probe_media(norm_silent_mp4) self.assertTrue(probe["has_audio"], "Output must retain audio track") self.assertEqual(probe["sample_rate"], 48000) self.assertEqual(probe["channels"], 2) self.__class__.telemetry_results["test_02_silent_resilience"] = { "measured_input_i": meas.get("input_i"), "measured_input_tp": meas.get("input_tp"), "is_silent_flag": meas.get("is_silent"), "normalization_success": ok, "output_channels": probe["channels"], "output_sample_rate": probe["sample_rate"] } # ------------------------------------------------------------------------- # TEST 3: Video Without Audio Stream (-an) Concatenation Resilience # ------------------------------------------------------------------------- def test_adv_03_missing_audio_stream_an_concat_resilience(self): """ Adversarial Test 3: Concatenating video without audio stream (-an) with standard audio video in various orderings does not crash and produces 48kHz stereo AAC. """ with tempfile.TemporaryDirectory() as td: v_audio = os.path.join(td, "clip_audio.mp4") v_mute = os.path.join(td, "clip_mute.mp4") v_mute2 = os.path.join(td, "clip_mute2.mp4") self._create_test_video(v_audio, duration=3.0, color="blue", has_audio=True) self._create_test_video(v_mute, duration=3.0, color="red", has_audio=False) self._create_test_video(v_mute2, duration=2.0, color="green", has_audio=False) sequences = [ {"name": "audio_then_mute", "clips": [v_audio, v_mute]}, {"name": "mute_then_audio", "clips": [v_mute, v_audio]}, {"name": "audio_mute_audio", "clips": [v_audio, v_mute, v_audio]}, {"name": "pure_mute_sequence", "clips": [v_mute, v_mute2]} ] telemetry_orderings = [] for seq in sequences: for mode in ["boundary_smoothing", "acrossfade"]: out_path = os.path.join(td, f"out_{seq['name']}_{mode}.mp4") ok = self.engine.stitch_with_audio_crossfade( seq["clips"], out_path, crossfade_dur=0.5, mode=mode ) self.assertTrue(ok, f"stitch_with_audio_crossfade failed for {seq['name']} in mode {mode}") self.assertTrue(os.path.exists(out_path)) probe = self._probe_media(out_path) self.assertTrue(probe["has_audio"], f"Output must have audio for {seq['name']} ({mode})") self.assertEqual(probe["audio_codec"], "aac") self.assertEqual(probe["sample_rate"], 48000, f"Sample rate must be 48000 for {seq['name']} ({mode})") self.assertEqual(probe["channels"], 2, f"Channels must be 2 (stereo) for {seq['name']} ({mode})") telemetry_orderings.append({ "sequence": seq["name"], "mode": mode, "success": ok, "codec": probe["audio_codec"], "sample_rate": probe["sample_rate"], "channels": probe["channels"], "format_duration": probe["format_duration"] }) self.__class__.telemetry_results["test_03_missing_audio_stream"] = telemetry_orderings # ------------------------------------------------------------------------- # TEST 4: Mixed Channel Counts (Mono + Stereo) Layout Preservation # ------------------------------------------------------------------------- def test_adv_04_mixed_channel_counts_stereo_preservation(self): """ Adversarial Test 4: Verify mixed channel inputs (mono + stereo) preserve stereo layout without downmixing to mono. """ with tempfile.TemporaryDirectory() as td: v_mono = os.path.join(td, "clip_mono.mp4") v_stereo = os.path.join(td, "clip_stereo.mp4") v_mono2 = os.path.join(td, "clip_mono2.mp4") self._create_test_video(v_mono, duration=3.0, channels=1, color="cyan") self._create_test_video(v_stereo, duration=3.0, channels=2, color="magenta") self._create_test_video(v_mono2, duration=2.5, channels=1, color="yellow") # Check inputs probe probe_m = self._probe_media(v_mono) probe_s = self._probe_media(v_stereo) self.assertEqual(probe_m["channels"], 1) self.assertEqual(probe_s["channels"], 2) test_cases = [ {"name": "mono_then_stereo", "clips": [v_mono, v_stereo]}, {"name": "stereo_then_mono", "clips": [v_stereo, v_mono]}, {"name": "mono_stereo_mono", "clips": [v_mono, v_stereo, v_mono2]} ] telemetry_channels = [] for tc in test_cases: out_a = os.path.join(td, f"out_{tc['name']}_mode_a.mp4") out_b = os.path.join(td, f"out_{tc['name']}_mode_b.mp4") # Mode A ok_a = self.engine.stitch_with_audio_crossfade(tc["clips"], out_a, mode="boundary_smoothing") self.assertTrue(ok_a) probe_out_a = self._probe_media(out_a) self.assertEqual(probe_out_a["channels"], 2, f"Mode A downmixed to {probe_out_a['channels']} for {tc['name']}") self.assertIn("stereo", probe_out_a["channel_layout"].lower()) self.assertEqual(probe_out_a["sample_rate"], 48000) # Mode B ok_b = self.engine.stitch_with_audio_crossfade(tc["clips"], out_b, mode="acrossfade", crossfade_dur=0.5) self.assertTrue(ok_b) probe_out_b = self._probe_media(out_b) self.assertEqual(probe_out_b["channels"], 2, f"Mode B downmixed to {probe_out_b['channels']} for {tc['name']}") self.assertIn("stereo", probe_out_b["channel_layout"].lower()) self.assertEqual(probe_out_b["sample_rate"], 48000) telemetry_channels.append({ "test_case": tc["name"], "mode_a_channels": probe_out_a["channels"], "mode_a_layout": probe_out_a["channel_layout"], "mode_b_channels": probe_out_b["channels"], "mode_b_layout": probe_out_b["channel_layout"] }) self.__class__.telemetry_results["test_04_mixed_channels"] = telemetry_channels # ------------------------------------------------------------------------- # TEST 5: Micro-fade Mode A Duration Preservation (Zero Timeline Shrinkage) # ------------------------------------------------------------------------- def test_adv_05_micro_fade_mode_a_duration_preservation(self): """ Adversarial Test 5: Verify Micro-fade Mode A (boundary_smoothing) preserves overall duration with 0.0s timeline loss across boundaries. """ with tempfile.TemporaryDirectory() as td: durations = [3.0, 4.0, 5.0] total_expected = sum(durations) # 12.0s clips = [] for idx, d in enumerate(durations): clip_p = os.path.join(td, f"clip_{idx}_{d}s.mp4") self._create_test_video(clip_p, duration=d, color=["red", "green", "blue"][idx]) clips.append(clip_p) out_mode_a = os.path.join(td, "out_mode_a_duration.mp4") ok = self.engine.stitch_with_audio_crossfade( clips, out_mode_a, crossfade_dur=1.0, mode="boundary_smoothing" ) self.assertTrue(ok, "Mode A stitching failed") probe = self._probe_media(out_mode_a) delta_v = abs(probe["video_duration"] - total_expected) delta_a = abs(probe["audio_duration"] - total_expected) delta_fmt = abs(probe["format_duration"] - total_expected) delta_va = abs(probe["video_duration"] - probe["audio_duration"]) # Tolerances: frame rate rounding <= 0.05s self.assertLessEqual( delta_v, 0.05, f"Video duration {probe['video_duration']:.3f} deviates from expected {total_expected:.3f}" ) self.assertLessEqual( delta_a, 0.05, f"Audio duration {probe['audio_duration']:.3f} deviates from expected {total_expected:.3f}" ) self.assertLessEqual( delta_fmt, 0.05, f"Format duration {probe['format_duration']:.3f} deviates from expected {total_expected:.3f}" ) self.assertLessEqual( delta_va, 0.05, f"Video/Audio duration delta {delta_va:.3f} exceeds 0.05s threshold" ) # Volumedetect sanity check on output (ensure no audio corruption) vol_report = self.engine.inspect_shot_audio(out_mode_a) self.assertGreater(vol_report["max_volume_db"], -50.0, "Audio should not be muted/corrupted") self.__class__.telemetry_results["test_05_mode_a_duration"] = { "expected_total_duration": total_expected, "actual_video_duration": probe["video_duration"], "actual_audio_duration": probe["audio_duration"], "actual_format_duration": probe["format_duration"], "delta_video": delta_v, "delta_audio": delta_a, "delta_av": delta_va, "zero_shrinkage_verified": delta_a <= 0.05 and delta_v <= 0.05 } # ------------------------------------------------------------------------- # TEST 6: Active Gain Staging Headroom Protection & Extreme Sample Rates # ------------------------------------------------------------------------- def test_adv_06_gain_staging_headroom_and_sample_rates(self): """ Adversarial Test 6: Verify active gain staging does not clip audio (respects -1.0 dBFS) and mismatched sample rates (22050, 44100, 48000) are unified without error. """ with tempfile.TemporaryDirectory() as td: # 1. Gain staging headroom test # Create a very loud clip with peak at -0.2 dB loud_clip = os.path.join(td, "hot_clip.mp4") self._create_test_video(loud_clip, duration=3.0, audio_filter="sine=f=440:r=48000:d=3,volume=-0.2dB[a_out]") normal_clip = os.path.join(td, "norm_clip.mp4") self._create_test_video(normal_clip, duration=3.0, audio_filter="sine=f=550:r=48000:d=3,volume=-18dB[a_out]") out_gain_staged = os.path.join(td, "out_staged.mp4") ok = self.engine.stitch_with_audio_crossfade( [loud_clip, normal_clip], out_gain_staged, mode="boundary_smoothing", active_gain_staging=True ) self.assertTrue(ok) staged_rep = self.engine.inspect_shot_audio(out_gain_staged) self.assertLessEqual( staged_rep["max_volume_db"], 0.0, f"Audio clipped: max volume is {staged_rep['max_volume_db']} dBFS" ) # 2. Extreme sample rates test (22050 Hz, 44100 Hz, 48000 Hz) c_22k = os.path.join(td, "c_22k.mp4") c_44k = os.path.join(td, "c_44k.mp4") c_48k = os.path.join(td, "c_48k.mp4") out_resampled = os.path.join(td, "out_resampled.mp4") self._create_test_video(c_22k, duration=2.0, sample_rate=22050) self._create_test_video(c_44k, duration=2.0, sample_rate=44100) self._create_test_video(c_48k, duration=2.0, sample_rate=48000) ok_resamp = self.engine.stitch_with_audio_crossfade( [c_22k, c_44k, c_48k], out_resampled, mode="boundary_smoothing" ) self.assertTrue(ok_resamp, "Stitching mismatched sample rates failed") probe_resamp = self._probe_media(out_resampled) self.assertEqual(probe_resamp["sample_rate"], 48000, "Final sample rate must be 48000 Hz") self.__class__.telemetry_results["test_06_gain_and_sample_rates"] = { "hot_clip_max_volume": staged_rep["max_volume_db"], "clipping_avoided": staged_rep["max_volume_db"] <= 0.0, "unified_sample_rate": probe_resamp["sample_rate"], "resample_success": ok_resamp } if __name__ == "__main__": unittest.main()