diff --git a/src/vocalinux/ui/audio_feedback.py b/src/vocalinux/ui/audio_feedback.py index e3536b6..1a8ab73 100644 --- a/src/vocalinux/ui/audio_feedback.py +++ b/src/vocalinux/ui/audio_feedback.py @@ -4,6 +4,7 @@ Audio feedback module for Vocalinux. This module provides audio feedback for various recognition states. """ +import hashlib import logging import os import shutil @@ -58,6 +59,14 @@ START_SOUND = _resource_manager.get_sound_path("start_recording") STOP_SOUND = _resource_manager.get_sound_path("stop_recording") ERROR_SOUND = _resource_manager.get_sound_path("error") +# Silent lead-in so a suspended PipeWire/WirePlumber sink can wake before the +# audible cue. Applied in the same WAV (one player process), not as a second +# stream. See #800. +_SINK_WAKE_PREROLL_MS = 100 +_REAL_AUDIO_PLAYERS = frozenset({"paplay", "aplay", "play", "mplayer"}) +_preroll_lock = threading.Lock() +_preroll_cache: dict[tuple[str, int, int], str] = {} + def _is_sound_effects_enabled() -> bool: try: @@ -92,6 +101,78 @@ def _wav_duration_seconds(sound_path: str) -> float: return 0.35 +def _pcm_silence(nframes: int, nchannels: int, sampwidth: int) -> bytes: + """Return PCM bytes of silence matching the source sample format.""" + if nframes < 1 or nchannels < 1 or sampwidth < 1: + raise ValueError("invalid PCM parameters for silence") + if sampwidth == 1: + # 8-bit WAV is unsigned; 0x80 is the zero point. + return bytes([0x80]) * (nframes * nchannels) + return b"\x00" * (nframes * nchannels * sampwidth) + + +def _preroll_cache_dir() -> str: + cache_home = os.environ.get("XDG_CACHE_HOME") or os.path.join(os.path.expanduser("~"), ".cache") + path = os.path.join(cache_home, "vocalinux", "audio-preroll") + os.makedirs(path, exist_ok=True) + return path + + +def _write_preroll_wav(source_path: str, dest_path: str, preroll_ms: int) -> None: + """Write ``source_path`` with ``preroll_ms`` of leading silence to ``dest_path``.""" + with wave.open(source_path, "rb") as src: + nchannels = src.getnchannels() + sampwidth = src.getsampwidth() + framerate = src.getframerate() + if nchannels < 1 or sampwidth < 1 or framerate < 1: + raise ValueError( + f"invalid WAV parameters: channels={nchannels} " + f"sampwidth={sampwidth} framerate={framerate}" + ) + audio = src.readframes(src.getnframes()) + preroll_frames = max(1, int(round(framerate * preroll_ms / 1000.0))) + silence = _pcm_silence(preroll_frames, nchannels, sampwidth) + with wave.open(dest_path, "wb") as dst: + dst.setnchannels(nchannels) + dst.setsampwidth(sampwidth) + dst.setframerate(framerate) + dst.writeframes(silence + audio) + + +def _prerolled_sound_path(sound_path: str, preroll_ms: int = _SINK_WAKE_PREROLL_MS) -> str: + """Return a cached WAV with silent preroll, or ``sound_path`` on failure.""" + try: + abs_path = os.path.abspath(sound_path) + mtime_ns = os.stat(abs_path).st_mtime_ns + key = (abs_path, mtime_ns, preroll_ms) + with _preroll_lock: + cached = _preroll_cache.get(key) + if cached is not None and os.path.isfile(cached): + return cached + digest = hashlib.sha256( + f"{abs_path}\0{mtime_ns}\0{preroll_ms}".encode("utf-8") + ).hexdigest() + dest = os.path.join(_preroll_cache_dir(), f"{digest[:16]}.wav") + if os.path.isfile(dest): + _preroll_cache[key] = dest + return dest + partial = dest + ".partial" + try: + _write_preroll_wav(abs_path, partial, preroll_ms) + os.replace(partial, dest) + except (OSError, ValueError, wave.Error, EOFError): + try: + os.unlink(partial) + except OSError: + pass + raise + _preroll_cache[key] = dest + return dest + except (OSError, ValueError, wave.Error, EOFError) as exc: + logger.warning("Could not prepend audio preroll for %s: %s", sound_path, exc) + return sound_path + + def _get_audio_player(): """ Determine the best available audio player on the system. @@ -126,7 +207,7 @@ def _get_audio_player(): return None, [] -def _play_sound_file(sound_path): +def _play_sound_file(sound_path: str) -> bool: """ Play a sound file using the best available player. @@ -158,31 +239,35 @@ def _play_sound_file(sound_path): logger.info(f"CI mode: Simulating playing sound {sound_path}") return True + playback_path = sound_path + if player in _REAL_AUDIO_PLAYERS: + playback_path = _prerolled_sound_path(sound_path) + try: if player == "paplay": subprocess.Popen( - [player, sound_path], + [player, playback_path], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, env=host_env(), ) elif player == "aplay": subprocess.Popen( - [player, "-q", sound_path], + [player, "-q", playback_path], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, env=host_env(), ) elif player == "mplayer": subprocess.Popen( - [player, "-really-quiet", sound_path], + [player, "-really-quiet", playback_path], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, env=host_env(), ) elif player == "play": subprocess.Popen( - [player, "-q", sound_path], + [player, "-q", playback_path], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, env=host_env(), diff --git a/tests/test_audio_feedback.py b/tests/test_audio_feedback.py index ba53b0d..1fd052a 100644 --- a/tests/test_audio_feedback.py +++ b/tests/test_audio_feedback.py @@ -4,7 +4,10 @@ Tests for the audio feedback functionality. import os import sys +import tempfile import unittest +import wave +from typing import Any from unittest.mock import patch import pytest @@ -12,20 +15,77 @@ import pytest # We need to use absolute paths for patching in module scope AUDIO_FEEDBACK_MODULE = "vocalinux.ui.audio_feedback" +# Pytest collects files in command-line order, so this module may import after +# tests that replace tempfile/wave in sys.modules with MagicMock. Rebind to +# the real stdlib modules for WAV fixtures. Leave real tempfile in sys.modules +# so a later `from tempfile import TemporaryDirectory` is not a MagicMock. +# Do not pop tempfile in the per-test fixture below. +if getattr(tempfile, "__name__", None) != "tempfile": + sys.modules.pop("tempfile", None) + import tempfile +if getattr(wave, "__name__", None) != "wave": + sys.modules.pop("wave", None) + import wave + @pytest.fixture(autouse=True) def reset_audio_module(): """Reset the audio_feedback module before each test to allow proper testing.""" + # Sibling tests replace sys.modules["wave"] with a MagicMock. Pin the real + # module this file imported so a reimported audio_feedback gets wave.open. + previous_wave = sys.modules.get("wave") + sys.modules["wave"] = wave + # Remove the mock that conftest installs if AUDIO_FEEDBACK_MODULE in sys.modules: del sys.modules[AUDIO_FEEDBACK_MODULE] - yield + try: + yield + finally: + if previous_wave is None: + sys.modules.pop("wave", None) + else: + sys.modules["wave"] = previous_wave - # Restore the mock after test for other tests that need it - from conftest import mock_audio_feedback + # Restore the mock after test for other tests that need it + from conftest import mock_audio_feedback - sys.modules[AUDIO_FEEDBACK_MODULE] = mock_audio_feedback + sys.modules[AUDIO_FEEDBACK_MODULE] = mock_audio_feedback + + +def _write_test_wav( + path: str, + *, + nframes: int = 80, + framerate: int = 8000, + nchannels: int = 1, + sampwidth: int = 2, + payload: bytes | None = None, +) -> str: + """Write a tiny PCM WAV fixture and return ``path``.""" + if payload is None: + payload = b"\x01\x00" * (nframes * nchannels) + with wave.open(path, "wb") as wav_file: + wav_file.setnchannels(nchannels) + wav_file.setsampwidth(sampwidth) + wav_file.setframerate(framerate) + wav_file.writeframes(payload) + return path + + +def _popen_argv(mock_popen: Any) -> list[str]: + args, _kwargs = mock_popen.call_args + return list(args[0]) + + +def _assert_played_wav(argv: list[str], player: str, extra_flags: tuple[str, ...] = ()) -> str: + """Assert player argv shape and return the wav path that was played.""" + assert argv[0] == player + assert argv[1 : 1 + len(extra_flags)] == list(extra_flags) + played = argv[-1] + assert str(played).endswith(".wav") + return played class TestAudioFeedback(unittest.TestCase): @@ -190,95 +250,84 @@ class TestAudioFeedback(unittest.TestCase): # Import the module first import vocalinux.ui.audio_feedback as audio_feedback - with ( - patch.object(audio_feedback.os.path, "exists", return_value=True), - patch.object( - audio_feedback, - "_get_audio_player", - return_value=("paplay", ["wav"]), - ), - patch.object(audio_feedback.subprocess, "Popen") as mock_popen, - ): - # Call the function - result = audio_feedback._play_sound_file("test.wav") + with tempfile.TemporaryDirectory() as tmp: + cue = _write_test_wav(os.path.join(tmp, "cue.wav")) + with ( + patch.object( + audio_feedback, + "_get_audio_player", + return_value=("paplay", ["wav"]), + ), + patch.object(audio_feedback.subprocess, "Popen") as mock_popen, + ): + result = audio_feedback._play_sound_file(cue) - # Verify the function returned True and called Popen correctly self.assertTrue(result) mock_popen.assert_called_once() - args, kwargs = mock_popen.call_args - self.assertEqual(args[0][0], "paplay") - self.assertEqual(args[0][1], "test.wav") + played = _assert_played_wav(_popen_argv(mock_popen), "paplay") + self.assertTrue(os.path.isfile(played), played) def test_play_sound_file_aplay(self): """Test playing sound with aplay.""" # Import the module first import vocalinux.ui.audio_feedback as audio_feedback - with ( - patch.object(audio_feedback.os.path, "exists", return_value=True), - patch.object( - audio_feedback, - "_get_audio_player", - return_value=("aplay", ["wav"]), - ), - patch.object(audio_feedback.subprocess, "Popen") as mock_popen, - ): - # Call the function - result = audio_feedback._play_sound_file("test.wav") + with tempfile.TemporaryDirectory() as tmp: + cue = _write_test_wav(os.path.join(tmp, "cue.wav")) + with ( + patch.object( + audio_feedback, + "_get_audio_player", + return_value=("aplay", ["wav"]), + ), + patch.object(audio_feedback.subprocess, "Popen") as mock_popen, + ): + result = audio_feedback._play_sound_file(cue) - # Verify the function returned True and called Popen correctly self.assertTrue(result) mock_popen.assert_called_once() - args, kwargs = mock_popen.call_args - self.assertEqual(args[0][0], "aplay") - self.assertEqual(args[0][1], "-q") - self.assertEqual(args[0][2], "test.wav") + played = _assert_played_wav(_popen_argv(mock_popen), "aplay", ("-q",)) + self.assertTrue(os.path.isfile(played), played) def test_play_sound_file_mplayer(self): """Test playing sound with mplayer.""" # Import the module first import vocalinux.ui.audio_feedback as audio_feedback - with ( - patch.object(audio_feedback.os.path, "exists", return_value=True), - patch.object( - audio_feedback, - "_get_audio_player", - return_value=("mplayer", ["wav"]), - ), - patch.object(audio_feedback.subprocess, "Popen") as mock_popen, - ): - # Call the function - result = audio_feedback._play_sound_file("test.wav") + with tempfile.TemporaryDirectory() as tmp: + cue = _write_test_wav(os.path.join(tmp, "cue.wav")) + with ( + patch.object( + audio_feedback, + "_get_audio_player", + return_value=("mplayer", ["wav"]), + ), + patch.object(audio_feedback.subprocess, "Popen") as mock_popen, + ): + result = audio_feedback._play_sound_file(cue) - # Verify the function returned True and called Popen correctly self.assertTrue(result) mock_popen.assert_called_once() - args, kwargs = mock_popen.call_args - self.assertEqual(args[0][0], "mplayer") - self.assertEqual(args[0][1], "-really-quiet") - self.assertEqual(args[0][2], "test.wav") + played = _assert_played_wav(_popen_argv(mock_popen), "mplayer", ("-really-quiet",)) + self.assertTrue(os.path.isfile(played), played) def test_play_sound_file_play(self): """Test playing sound with play (SoX).""" # Import the module first import vocalinux.ui.audio_feedback as audio_feedback - with ( - patch.object(audio_feedback.os.path, "exists", return_value=True), - patch.object(audio_feedback, "_get_audio_player", return_value=("play", ["wav"])), - patch.object(audio_feedback.subprocess, "Popen") as mock_popen, - ): - # Call the function - result = audio_feedback._play_sound_file("test.wav") + with tempfile.TemporaryDirectory() as tmp: + cue = _write_test_wav(os.path.join(tmp, "cue.wav")) + with ( + patch.object(audio_feedback, "_get_audio_player", return_value=("play", ["wav"])), + patch.object(audio_feedback.subprocess, "Popen") as mock_popen, + ): + result = audio_feedback._play_sound_file(cue) - # Verify the function returned True and called Popen correctly self.assertTrue(result) mock_popen.assert_called_once() - args, kwargs = mock_popen.call_args - self.assertEqual(args[0][0], "play") - self.assertEqual(args[0][1], "-q") - self.assertEqual(args[0][2], "test.wav") + played = _assert_played_wav(_popen_argv(mock_popen), "play", ("-q",)) + self.assertTrue(os.path.isfile(played), played) def test_play_sound_file_exception(self): """Test handling exception when playing sound.""" @@ -304,6 +353,124 @@ class TestAudioFeedback(unittest.TestCase): # Verify the function returned False self.assertFalse(result) + def test_play_sound_file_prepends_silent_preroll(self): + """Real players play a WAV with leading silence of the preroll duration.""" + import vocalinux.ui.audio_feedback as audio_feedback + + framerate = 8000 + nchannels = 1 + sampwidth = 2 + nframes = 80 + payload = b"\x01\x00" * nframes + + with tempfile.TemporaryDirectory() as tmp: + cue = _write_test_wav( + os.path.join(tmp, "cue.wav"), + nframes=nframes, + framerate=framerate, + nchannels=nchannels, + sampwidth=sampwidth, + payload=payload, + ) + with ( + patch.object( + audio_feedback, + "_get_audio_player", + return_value=("paplay", ["wav"]), + ), + patch.object(audio_feedback.subprocess, "Popen") as mock_popen, + ): + result = audio_feedback._play_sound_file(cue) + + self.assertTrue(result) + mock_popen.assert_called_once() + played = _assert_played_wav(_popen_argv(mock_popen), "paplay") + self.assertTrue(os.path.isfile(played), played) + self.assertNotEqual(os.path.realpath(played), os.path.realpath(cue)) + + with wave.open(cue, "rb") as src, wave.open(played, "rb") as dst: + self.assertEqual(dst.getnchannels(), nchannels) + self.assertEqual(dst.getsampwidth(), sampwidth) + self.assertEqual(dst.getframerate(), framerate) + src_frames = src.getnframes() + dst_frames = dst.getnframes() + dst_bytes = dst.readframes(dst_frames) + + preroll_ms = audio_feedback._SINK_WAKE_PREROLL_MS + expected_silence = max(1, int(round(framerate * preroll_ms / 1000.0))) + self.assertEqual(dst_frames, src_frames + expected_silence) + + frame_size = nchannels * sampwidth + silence = dst_bytes[: expected_silence * frame_size] + rest = dst_bytes[expected_silence * frame_size :] + self.assertEqual(silence, b"\x00" * len(silence)) + self.assertEqual(rest, payload) + + def test_preroll_cache_reuses_built_file(self): + """The same source path and mtime reuse the prerolled WAV.""" + import vocalinux.ui.audio_feedback as audio_feedback + + with tempfile.TemporaryDirectory() as tmp: + cue = _write_test_wav(os.path.join(tmp, "cue.wav")) + with ( + patch.object( + audio_feedback, + "_get_audio_player", + return_value=("paplay", ["wav"]), + ), + patch.object(audio_feedback.subprocess, "Popen") as mock_popen, + ): + self.assertTrue(audio_feedback._play_sound_file(cue)) + self.assertTrue(audio_feedback._play_sound_file(cue)) + + self.assertEqual(mock_popen.call_count, 2) + first = _assert_played_wav(list(mock_popen.call_args_list[0].args[0]), "paplay") + second = _assert_played_wav(list(mock_popen.call_args_list[1].args[0]), "paplay") + self.assertEqual(first, second) + + def test_preroll_build_failure_falls_back_to_original(self): + """A preroll build error plays the original cue instead of failing closed.""" + import vocalinux.ui.audio_feedback as audio_feedback + + with tempfile.TemporaryDirectory() as tmp: + cue = _write_test_wav(os.path.join(tmp, "cue.wav")) + with ( + patch.object( + audio_feedback, + "_get_audio_player", + return_value=("paplay", ["wav"]), + ), + patch.object(audio_feedback.wave, "open", side_effect=wave.Error("boom")), + patch.object(audio_feedback.subprocess, "Popen") as mock_popen, + ): + result = audio_feedback._play_sound_file(cue) + + self.assertTrue(result) + played = _assert_played_wav(_popen_argv(mock_popen), "paplay") + self.assertEqual(os.path.realpath(played), os.path.realpath(cue)) + + def test_preroll_eoferror_falls_back_to_original(self): + """A truncated/corrupt WAV raising EOFError still plays the original cue.""" + import vocalinux.ui.audio_feedback as audio_feedback + + with tempfile.TemporaryDirectory() as tmp: + cue = os.path.join(tmp, "cue.wav") + with open(cue, "wb") as handle: + handle.write(b"RIFF") + with ( + patch.object( + audio_feedback, + "_get_audio_player", + return_value=("paplay", ["wav"]), + ), + patch.object(audio_feedback.subprocess, "Popen") as mock_popen, + ): + result = audio_feedback._play_sound_file(cue) + + self.assertTrue(result) + played = _assert_played_wav(_popen_argv(mock_popen), "paplay") + self.assertEqual(os.path.realpath(played), os.path.realpath(cue)) + def test_play_start_sound(self): """Test playing start sound (default tone is voca).""" # Import the module first @@ -347,24 +514,23 @@ class TestAudioFeedback(unittest.TestCase): # Import the module first import vocalinux.ui.audio_feedback as audio_feedback - with ( - patch.object(audio_feedback.os.path, "exists", return_value=True), - patch.object( - audio_feedback, - "_get_audio_player", - return_value=("ci_test_player", ["wav"]), - ), - patch.object(audio_feedback.subprocess, "Popen") as mock_popen, - ): - # Call the function - result = audio_feedback._play_sound_file("test.wav") + with tempfile.TemporaryDirectory() as tmp: + cue = _write_test_wav(os.path.join(tmp, "cue.wav")) + with ( + patch.object( + audio_feedback, + "_get_audio_player", + return_value=("ci_test_player", ["wav"]), + ), + patch.object(audio_feedback.subprocess, "Popen") as mock_popen, + ): + result = audio_feedback._play_sound_file(cue) - # Verify the function returned True and called Popen correctly self.assertTrue(result) mock_popen.assert_called_once() - args, kwargs = mock_popen.call_args - self.assertEqual(args[0][0], "ci_test_player") - self.assertEqual(args[0][1], "test.wav") + argv = _popen_argv(mock_popen) + self.assertEqual(argv[0], "ci_test_player") + self.assertEqual(argv[1], cue) def test_get_audio_player_github_actions_fallback(self): """Test ci_test_player assignment in GitHub Actions without audio player.""" diff --git a/tests/test_whisper_model_info.py b/tests/test_whisper_model_info.py index 893aee6..84d4c25 100644 --- a/tests/test_whisper_model_info.py +++ b/tests/test_whisper_model_info.py @@ -8,11 +8,19 @@ dialog that reported "large" as missing no matter how often it was fetched. """ import os +import sys import unittest from pathlib import Path from tempfile import TemporaryDirectory from unittest.mock import patch +# test_recognition_manager / test_speech_recognition replace sys.modules +# ["tempfile"] with a MagicMock at import time and never put it back. Bind +# TemporaryDirectory from the real stdlib module even when that happens. +if not isinstance(TemporaryDirectory, type): + sys.modules.pop("tempfile", None) + from tempfile import TemporaryDirectory + from vocalinux.utils.whisper_model_info import ( WHISPER_MODEL_SIZES, migrate_legacy_checkpoint_names,