import subprocess import tempfile import unittest import wave from pathlib import Path from unittest.mock import patch from src.meeting_lab.audio.preparation import ( DEFAULT_NORMALIZATION_FILTER, DEFAULT_NORMALIZATION_METHOD, AudioPreparationError, prepare_audio, ) def _write_wav( path: Path, *, channels: int = 1, sample_rate: int = 16_000, sample_width: int = 2 ) -> None: with wave.open(str(path), "wb") as recording: recording.setnchannels(channels) recording.setsampwidth(sample_width) recording.setframerate(sample_rate) recording.writeframes(b"\x00" * channels * sample_width * 32) def _successful_runner(commands: list[list[str]]): def run(command, **kwargs): commands.append(list(command)) _write_wav(Path(command[-1])) return subprocess.CompletedProcess(command, 0, "", "") return run class AudioPreparationTests(unittest.TestCase): def test_supported_inputs_are_prepared_with_normalization_on_and_off(self) -> None: for suffix in (".wav", ".flac", ".m4a"): for normalization_enabled in (True, False): with ( self.subTest( suffix=suffix, normalization_enabled=normalization_enabled ), tempfile.TemporaryDirectory() as directory, ): root = Path(directory) source = root / f"meeting{suffix}" if suffix == ".wav": _write_wav(source) else: source.write_bytes(b"original encoded audio") original = source.read_bytes() destination = root / "run" / "audio" / "prepared.wav" commands: list[list[str]] = [] with patch( "src.meeting_lab.audio.preparation.shutil.which", return_value="/usr/bin/ffmpeg", ): result = prepare_audio( source, destination, normalization_enabled=normalization_enabled, runner=_successful_runner(commands), ) self.assertEqual(source.read_bytes(), original) self.assertEqual(result.prepared_path, destination) with wave.open(str(destination), "rb") as recording: self.assertEqual(recording.getnchannels(), 1) self.assertEqual(recording.getframerate(), 16_000) self.assertEqual(recording.getsampwidth(), 2) self.assertEqual(recording.getcomptype(), "NONE") self.assertEqual(commands[0][commands[0].index("-ac") + 1], "1") self.assertEqual(commands[0][commands[0].index("-ar") + 1], "16000") self.assertEqual( commands[0][commands[0].index("-c:a") + 1], "pcm_s16le" ) self.assertEqual("-af" in commands[0], normalization_enabled) if normalization_enabled: self.assertEqual( commands[0][commands[0].index("-af") + 1], DEFAULT_NORMALIZATION_FILTER, ) self.assertEqual( result.normalization_enabled, normalization_enabled ) def test_normalization_defaults_to_on_and_explicit_on_matches(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) source = root / "meeting.wav" _write_wav(source) commands: list[list[str]] = [] with patch( "src.meeting_lab.audio.preparation.shutil.which", return_value="/usr/bin/ffmpeg", ): default = prepare_audio( source, root / "default.wav", runner=_successful_runner(commands) ) explicit = prepare_audio( source, root / "explicit.wav", normalization_enabled=True, runner=_successful_runner(commands), ) self.assertTrue(default.normalization_enabled) self.assertTrue(explicit.normalization_enabled) self.assertEqual( commands[0][commands[0].index("-af") + 1], commands[1][commands[1].index("-af") + 1], ) def test_noncanonical_wav_is_normalized(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) source = root / "stereo-48k.wav" _write_wav(source, channels=2, sample_rate=48_000) destination = root / "prepared.wav" with patch( "src.meeting_lab.audio.preparation.shutil.which", return_value="/usr/bin/ffmpeg", ): prepare_audio(source, destination, runner=_successful_runner([])) with wave.open(str(destination), "rb") as recording: self.assertEqual( (recording.getnchannels(), recording.getframerate()), (1, 16_000) ) def test_ffmpeg_missing_has_actionable_error(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) source = root / "meeting.flac" source.write_bytes(b"audio") with ( patch( "src.meeting_lab.audio.preparation.shutil.which", return_value=None ), self.assertRaisesRegex(AudioPreparationError, "not found on PATH"), ): prepare_audio(source, root / "prepared.wav") def test_ffmpeg_failure_includes_diagnostic_and_preserves_source(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) source = root / "meeting.m4a" source.write_bytes(b"original") def fail(command, **kwargs): return subprocess.CompletedProcess(command, 1, "", "decoder exploded") with ( patch( "src.meeting_lab.audio.preparation.shutil.which", return_value="/usr/bin/ffmpeg", ), self.assertRaisesRegex(AudioPreparationError, "decoder exploded"), ): prepare_audio(source, root / "prepared.wav", runner=fail) self.assertEqual(source.read_bytes(), b"original") self.assertFalse((root / "prepared.wav").exists()) def test_prepared_audio_metadata_is_traceable(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) source = root / "unknown_meeting.flac" source.write_bytes(b"source") destination = root / "audio" / "prepared.wav" with patch( "src.meeting_lab.audio.preparation.shutil.which", return_value="/usr/bin/ffmpeg", ): result = prepare_audio( source, destination, runner=_successful_runner([]) ) metadata = result.metadata() self.assertEqual(metadata["original_source_name"], "unknown_meeting.flac") self.assertEqual(metadata["original_format"], "flac") self.assertEqual( metadata["prepared_audio_path"], str(destination.resolve()) ) self.assertEqual(metadata["preparation_method"], "ffmpeg") self.assertTrue(metadata["normalization_enabled"]) self.assertEqual( metadata["normalization_method"], DEFAULT_NORMALIZATION_METHOD ) self.assertEqual( metadata["normalization_filter"], DEFAULT_NORMALIZATION_FILTER ) self.assertEqual( metadata["canonical_output"], { "container": "wav", "codec": "pcm_s16le", "channels": 1, "sample_rate_hz": 16_000, "bits_per_sample": 16, }, ) def test_disabled_normalization_metadata_has_no_method_or_filter(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) source = root / "meeting.m4a" source.write_bytes(b"source") with patch( "src.meeting_lab.audio.preparation.shutil.which", return_value="/usr/bin/ffmpeg", ): result = prepare_audio( source, root / "prepared.wav", normalization_enabled=False, runner=_successful_runner([]), ) metadata = result.metadata() self.assertFalse(metadata["normalization_enabled"]) self.assertIsNone(metadata["normalization_method"]) self.assertIsNone(metadata["normalization_filter"]) if __name__ == "__main__": unittest.main()