* fix(config): share one ConfigManager across the process main(), the tray, and the sound-effects check each constructed their own ConfigManager. Every instance loads the whole config into memory at construction time and save_config() writes that whole cache back, so any save from a stale instance silently reverts what another instance saved in the meantime — no race needed, the second save just wins with old values. The disk-read workarounds in main (_should_append_trailing_space) already existed to dodge exactly this. Add get_shared_config_manager() and route the three call sites through it. It keeps ConfigManager constructible for tests, and the tray/dialog pair already shared one instance, so this extends that arrangement to the rest of the process. Tests: a regression test runs the two-instance scenario through the accessor, and a source guard fails if application code constructs ConfigManager directly again. conftest drops the shared instance between tests, since it would otherwise be created under whichever mocks the first caller had active and leak them into later tests. The two single-instance tests now patch acquire_lock itself: they used patch.dict(sys.modules, ...), which main()'s "from . import" ignores once an earlier test has imported the real module — they only passed through an unrelated mock leaking into the tray's shortcut setup, which this change stops. * fix(config): lock the shared accessor and prove the revert it prevents Review follow-ups on #691: - guard the lazy create with a lock, so two threads racing the first call cannot each build a ConfigManager and reintroduce #689 - take both handles before the first save in the accessor test, and assert identity there — the old order let a fresh ConfigManager pass - add a characterization test: two direct ConfigManager instances still revert each other, which is the footgun the accessor exists to hide - patch get_shared_config_manager in test_main and test_audio_feedback instead of the class, so a leftover singleton cannot skip the mock - close the ExitStack in _boot_under_patches when main() raises; the caller only closes a stack it received, so a failure there leaked those patches into every later test * fix(audio): route the tone lookup through the shared ConfigManager The tone picker merged on main (#707) constructs its own ConfigManager, which is exactly the stale-cache overwrite this branch removes; the new source guard in test_shared_config_manager.py flags it.
231 lines
7.0 KiB
Python
231 lines
7.0 KiB
Python
"""
|
|
Audio feedback module for Vocalinux.
|
|
|
|
This module provides audio feedback for various recognition states.
|
|
"""
|
|
|
|
import logging
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import wave
|
|
from pathlib import Path # noqa: F401
|
|
from typing import Optional
|
|
|
|
from .config_manager import DEFAULT_SOUND_EFFECT_TONE, normalize_sound_effect_tone
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Set a flag for CI/test environments
|
|
# This will be used to make sound functions work in CI testing environments
|
|
# Only use mock player in CI when not explicitly testing the player detection
|
|
|
|
|
|
def _is_ci_mode():
|
|
"""Check if we're in CI mode and should use mock audio player.
|
|
|
|
Returns False when running under pytest to allow proper unit testing.
|
|
"""
|
|
# If not in GitHub Actions, definitely not CI mode
|
|
if os.environ.get("GITHUB_ACTIONS") != "true":
|
|
return False
|
|
|
|
# Check for pytest in multiple ways to be comprehensive
|
|
# When running pytest, we want to return False so tests can properly
|
|
# mock and test the audio player detection logic
|
|
running_pytest = (
|
|
"pytest" in sys.modules
|
|
or "_pytest" in sys.modules
|
|
or "PYTEST_CURRENT_TEST" in os.environ
|
|
or any("pytest" in arg or arg.endswith("pytest") for arg in sys.argv)
|
|
or os.environ.get("PYTEST_RUNNING") == "1"
|
|
)
|
|
return not running_pytest
|
|
|
|
|
|
# Import the centralized resource manager
|
|
from ..utils.resource_manager import ResourceManager # noqa: E402
|
|
|
|
# Initialize resource manager
|
|
_resource_manager = ResourceManager()
|
|
|
|
# Sound file paths
|
|
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")
|
|
|
|
|
|
def _is_sound_effects_enabled() -> bool:
|
|
try:
|
|
from .config_manager import get_shared_config_manager
|
|
|
|
return get_shared_config_manager().is_sound_effects_enabled()
|
|
except Exception:
|
|
return True
|
|
|
|
|
|
def _resolved_tone() -> str:
|
|
try:
|
|
from .config_manager import get_shared_config_manager
|
|
|
|
return get_shared_config_manager().get_sound_effects_tone()
|
|
except Exception:
|
|
return DEFAULT_SOUND_EFFECT_TONE
|
|
|
|
|
|
def tone_sound_path(tone_id: str, kind: str) -> str:
|
|
"""Return the WAV path for a catalog start/stop cue."""
|
|
return _resource_manager.get_sound_path(f"{tone_id}_{kind}")
|
|
|
|
|
|
def _wav_duration_seconds(sound_path: str) -> float:
|
|
try:
|
|
with wave.open(sound_path, "rb") as wav_file:
|
|
frames = wav_file.getnframes()
|
|
rate = wav_file.getframerate() or 1
|
|
return frames / float(rate)
|
|
except Exception:
|
|
return 0.35
|
|
|
|
|
|
def _get_audio_player():
|
|
"""
|
|
Determine the best available audio player on the system.
|
|
|
|
Returns:
|
|
tuple: (player_command, supported_formats)
|
|
"""
|
|
# In CI mode, return a mock player to make tests pass,
|
|
# but only when not running pytest (to avoid interfering with unit tests)
|
|
if _is_ci_mode():
|
|
logger.info("CI mode: Using mock audio player")
|
|
return "mock_player", ["wav"]
|
|
|
|
# Check for PulseAudio paplay (preferred)
|
|
if shutil.which("paplay"):
|
|
return "paplay", ["wav"]
|
|
|
|
# Check for ALSA aplay
|
|
if shutil.which("aplay"):
|
|
return "aplay", ["wav"]
|
|
|
|
# Check for play (from SoX)
|
|
if shutil.which("play"):
|
|
return "play", ["wav"]
|
|
|
|
# Check for mplayer
|
|
if shutil.which("mplayer"):
|
|
return "mplayer", ["wav"]
|
|
|
|
# No suitable player found
|
|
logger.warning("No suitable audio player found for sound notifications")
|
|
return None, []
|
|
|
|
|
|
def _play_sound_file(sound_path):
|
|
"""
|
|
Play a sound file using the best available player.
|
|
|
|
Args:
|
|
sound_path: Path to the sound file
|
|
|
|
Returns:
|
|
bool: True if sound was played successfully, False otherwise
|
|
"""
|
|
if not os.path.exists(sound_path):
|
|
logger.warning(f"Sound file not found: {sound_path}")
|
|
return False
|
|
|
|
player, formats = _get_audio_player()
|
|
|
|
# Special handling for CI environment during tests
|
|
# If we're in CI (no audio players available) but running tests,
|
|
# continue with the execution to allow proper mocking
|
|
if not player and os.environ.get("GITHUB_ACTIONS") == "true":
|
|
# In CI tests with no audio player, use a placeholder to allow mocking to work
|
|
player = "ci_test_player"
|
|
|
|
if not player:
|
|
return False
|
|
|
|
# In CI mode, just pretend we played the sound and return success
|
|
# but only when not running pytest (to avoid interfering with unit tests)
|
|
if _is_ci_mode() and player == "mock_player":
|
|
logger.info(f"CI mode: Simulating playing sound {sound_path}")
|
|
return True
|
|
|
|
try:
|
|
if player == "paplay":
|
|
subprocess.Popen(
|
|
[player, sound_path],
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
)
|
|
elif player == "aplay":
|
|
subprocess.Popen(
|
|
[player, "-q", sound_path],
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
)
|
|
elif player == "mplayer":
|
|
subprocess.Popen(
|
|
[player, "-really-quiet", sound_path],
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
)
|
|
elif player == "play":
|
|
subprocess.Popen(
|
|
[player, "-q", sound_path],
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
)
|
|
elif player == "ci_test_player":
|
|
# This is a placeholder for CI tests - the subprocess call will be mocked
|
|
subprocess.Popen(
|
|
["ci_test_player", sound_path],
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
)
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"Failed to play sound {sound_path}: {e}")
|
|
return False
|
|
|
|
|
|
def play_start_sound():
|
|
if not _is_sound_effects_enabled():
|
|
return False
|
|
tone = _resolved_tone()
|
|
if tone == "off":
|
|
return False
|
|
return _play_sound_file(tone_sound_path(tone, "start"))
|
|
|
|
|
|
def play_stop_sound():
|
|
if not _is_sound_effects_enabled():
|
|
return False
|
|
tone = _resolved_tone()
|
|
if tone == "off":
|
|
return False
|
|
return _play_sound_file(tone_sound_path(tone, "stop"))
|
|
|
|
|
|
def play_error_sound():
|
|
return _play_sound_file(ERROR_SOUND) if _is_sound_effects_enabled() else False
|
|
|
|
|
|
def preview_tone(tone_id: Optional[str] = None) -> bool:
|
|
"""Play start then stop for a catalog tone. Off plays nothing."""
|
|
tone = normalize_sound_effect_tone(tone_id if tone_id is not None else _resolved_tone())
|
|
if tone == "off":
|
|
return False
|
|
start_path = tone_sound_path(tone, "start")
|
|
stop_path = tone_sound_path(tone, "stop")
|
|
if not _play_sound_file(start_path):
|
|
return False
|
|
delay = _wav_duration_seconds(start_path) + 0.08
|
|
threading.Timer(delay, _play_sound_file, args=(stop_path,)).start()
|
|
return True
|