231 lines
9.2 KiB
Python
231 lines
9.2 KiB
Python
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()
|