Add canonical audio preparation and meeting context support
This commit is contained in:
@@ -0,0 +1,230 @@
|
||||
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()
|
||||
Reference in New Issue
Block a user