Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
100 changes: 100 additions & 0 deletions tests/tools/test_tts_macos_output.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,100 @@
"""macOS output policy for streaming TTS.

On macOS, stream_tts_to_speaker must NOT open a sounddevice OutputStream
(PortAudio/CoreAudio init triggers a kTCCServiceMediaLibrary prompt). It
should route audio through the tempfile/afplay fallback instead.
See PR #62601 / #13291.
"""

import queue
import threading

import pytest


def _run_stream(monkeypatch, system_name):
"""Drive stream_tts_to_speaker once with a mock client on *system_name*.

Returns True if _import_sounddevice was called during the run.
"""
import tools.tts_tool as tts

monkeypatch.setattr("tools.tts_tool.platform.system", lambda: system_name)
monkeypatch.setattr("tools.tts_tool.get_env_value",
lambda name, default=None: "fake-key"
if name == "ELEVENLABS_API_KEY" else default)
monkeypatch.setattr("tools.tts_tool._load_tts_config", lambda: {})

class _FakeTTS:
def __init__(self, *a, **k):
self.text_to_speech = self

def convert(self, *a, **k):
return iter([]) # no audio chunks needed for setup assertion

monkeypatch.setattr("tools.tts_tool._import_elevenlabs", lambda: _FakeTTS)

sd_called = {"hit": False}

def _spy_import_sd():
sd_called["hit"] = True
raise AssertionError("sounddevice must not be imported for output on macOS")

monkeypatch.setattr("tools.tts_tool._import_sounddevice", _spy_import_sd)

text_queue: queue.Queue = queue.Queue()
text_queue.put(None) # end-of-text sentinel: no sentence spoken
stop_event = threading.Event()
done_event = threading.Event()

tts.stream_tts_to_speaker(text_queue, stop_event, done_event)
assert done_event.is_set()
return sd_called["hit"]


def test_streaming_tts_skips_sounddevice_on_macos(monkeypatch):
assert _run_stream(monkeypatch, "Darwin") is False


def test_streaming_tts_uses_sounddevice_off_macos(monkeypatch):
# Off macOS the OutputStream setup runs; _import_sounddevice raising here
# is caught by the function's own guard, so the call itself is what we assert.
called = _run_stream_offmac(monkeypatch)
assert called is True


def _run_stream_offmac(monkeypatch):
"""Like _run_stream but tolerant of the sounddevice import being attempted."""
import tools.tts_tool as tts

monkeypatch.setattr("tools.tts_tool.platform.system", lambda: "Linux")
monkeypatch.setattr("tools.tts_tool.get_env_value",
lambda name, default=None: "fake-key"
if name == "ELEVENLABS_API_KEY" else default)
monkeypatch.setattr("tools.tts_tool._load_tts_config", lambda: {})

class _FakeTTS:
def __init__(self, *a, **k):
self.text_to_speech = self

def convert(self, *a, **k):
return iter([])

monkeypatch.setattr("tools.tts_tool._import_elevenlabs", lambda: _FakeTTS)

sd_called = {"hit": False}

def _spy_import_sd():
sd_called["hit"] = True
raise OSError("no audio device in test") # handled by the function's guard

monkeypatch.setattr("tools.tts_tool._import_sounddevice", _spy_import_sd)

text_queue: queue.Queue = queue.Queue()
text_queue.put(None)
stop_event = threading.Event()
done_event = threading.Event()

tts.stream_tts_to_speaker(text_queue, stop_event, done_event)
assert done_event.is_set()
return sd_called["hit"]
104 changes: 104 additions & 0 deletions tests/tools/test_voice_mode.py
Original file line number Diff line number Diff line change
Expand Up @@ -883,6 +883,10 @@ def test_real_speech_not_filtered(self):
class TestPlayAudioFile:
def test_play_wav_via_sounddevice(self, monkeypatch, sample_wav):
np = pytest.importorskip("numpy")
# Pin to a non-macOS platform: on macOS WAV output deliberately skips
# sounddevice (see TestMacOSAudioOutputPolicy), so this path is only
# exercised off Darwin.
monkeypatch.setattr("tools.voice_mode.platform.system", lambda: "Linux")

mock_sd_obj = MagicMock()
# Simulate stream completing immediately (get_stream().active = False)
Expand Down Expand Up @@ -921,6 +925,106 @@ def test_returns_false_for_missing_file(self):
assert result is False


# ============================================================================
# macOS output policy (no sounddevice for OUTPUT -> avoids TCC prompt)
# ============================================================================

class TestMacOSAudioOutputPolicy:
def test_output_disallowed_on_macos(self, monkeypatch):
monkeypatch.setattr("tools.voice_mode.platform.system", lambda: "Darwin")
from tools.voice_mode import _sounddevice_output_allowed

assert _sounddevice_output_allowed() is False

def test_output_allowed_off_macos(self, monkeypatch):
monkeypatch.setattr("tools.voice_mode.platform.system", lambda: "Linux")
from tools.voice_mode import _sounddevice_output_allowed

assert _sounddevice_output_allowed() is True

def test_play_audio_file_skips_sounddevice_on_macos(self, monkeypatch, sample_wav):
"""On macOS, WAV playback must not import sounddevice; it routes to afplay."""
monkeypatch.setattr("tools.voice_mode.platform.system", lambda: "Darwin")

def _forbidden_import():
raise AssertionError("sounddevice must not be imported for output on macOS")

monkeypatch.setattr("tools.voice_mode._import_audio", _forbidden_import)

popen_cmds = []

class _FakeProc:
def wait(self, timeout=None):
return 0

def kill(self):
pass

def _fake_popen(cmd, **kwargs):
popen_cmds.append(cmd)
return _FakeProc()

monkeypatch.setattr("shutil.which", lambda exe: f"/usr/bin/{exe}")
monkeypatch.setattr("subprocess.Popen", _fake_popen)

from tools.voice_mode import play_audio_file

result = play_audio_file(sample_wav)

assert result is True
assert popen_cmds, "expected a system player to be invoked"
assert popen_cmds[0][0] == "afplay"

def test_play_beep_routes_through_afplay_on_macos(self, monkeypatch):
"""On macOS, beeps synthesize with numpy but play via the tempfile/afplay path."""
pytest.importorskip("numpy")
monkeypatch.setattr("tools.voice_mode.platform.system", lambda: "Darwin")

def _forbidden_import():
raise AssertionError("sounddevice must not be imported for beeps on macOS")

monkeypatch.setattr("tools.voice_mode._import_audio", _forbidden_import)

calls = []
monkeypatch.setattr(
"tools.voice_mode._play_int16_via_tempfile",
lambda audio, sample_rate: calls.append((len(audio), sample_rate)),
)

import tools.voice_mode as vm

vm.play_beep(frequency=880, count=1)

assert len(calls) == 1
n_samples, sample_rate = calls[0]
assert n_samples > 0
assert sample_rate == vm.SAMPLE_RATE

def test_play_beep_uses_sounddevice_off_macos(self, monkeypatch):
"""Off macOS, beeps go straight through sounddevice."""
np = pytest.importorskip("numpy")
monkeypatch.setattr("tools.voice_mode.platform.system", lambda: "Linux")

mock_sd = MagicMock()
mock_stream = MagicMock()
mock_stream.active = False
mock_sd.get_stream.return_value = mock_stream
monkeypatch.setattr("tools.voice_mode._import_audio", lambda: (mock_sd, np))

tempfile_calls = []
monkeypatch.setattr(
"tools.voice_mode._play_int16_via_tempfile",
lambda audio, sample_rate: tempfile_calls.append(True),
)

import tools.voice_mode as vm

vm.play_beep(frequency=880, count=1)

mock_sd.play.assert_called_once()
assert not tempfile_calls, "off macOS should not use the tempfile/afplay path"


# ============================================================================
# cleanup_temp_recordings
# ============================================================================
Expand Down
9 changes: 8 additions & 1 deletion tools/tts_tool.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@
import logging
import os
import queue
import platform
import re
import shlex
import shutil
Expand Down Expand Up @@ -2645,7 +2646,13 @@ def stream_tts_to_speaker(
# Open a single sounddevice output stream for the lifetime of
# this function. ElevenLabs pcm_24000 produces signed 16-bit
# little-endian mono PCM at 24 kHz.
if client is not None:
#
# On macOS, skip the sounddevice OutputStream entirely: PortAudio/
# CoreAudio init triggers a kTCCServiceMediaLibrary permission
# prompt even though output needs no media-library access. Leaving
# output_stream=None routes each sentence through _play_via_tempfile
# -> play_audio_file -> afplay. See PR #62601 / #13291.
if client is not None and platform.system() != "Darwin":
try:
sd = _import_sounddevice()
output_stream = sd.OutputStream(
Expand Down
72 changes: 68 additions & 4 deletions tools/voice_mode.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,55 @@ def _import_audio():
return sd, np


def _import_numpy():
"""Lazy-import numpy only (no sounddevice). Returns the module.

Used where we need to synthesize/convert audio samples but must NOT
import sounddevice — see _sounddevice_output_allowed.
"""
import numpy as np
return np


def _sounddevice_output_allowed() -> bool:
"""Whether sounddevice may be used for audio OUTPUT.

Returns False on macOS: importing/initializing sounddevice
(PortAudio/CoreAudio) for output triggers a kTCCServiceMediaLibrary
permission prompt, even though playback needs no media-library access.
On macOS all output is routed through ``afplay`` instead. This does NOT
affect audio *input* (recording), which legitimately needs microphone
permission. See PR #62601 / #13291.
"""
return platform.system() != "Darwin"


def _play_int16_via_tempfile(audio, sample_rate: int) -> None:
"""Write int16 mono PCM to a temp WAV and play it via play_audio_file.

Used on macOS so tone/beep output goes through ``afplay`` instead of
sounddevice (avoids the TCC media-library prompt).
"""
tmp_path = None
try:
tmp = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
tmp_path = tmp.name
with wave.open(tmp, "wb") as wf:
wf.setnchannels(1)
wf.setsampwidth(2) # 16-bit
wf.setframerate(sample_rate)
wf.writeframes(audio.tobytes())
play_audio_file(tmp_path)
except Exception as e:
logger.debug("Tone tempfile playback failed: %s", e)
finally:
if tmp_path:
try:
os.unlink(tmp_path)
except OSError:
pass


def _audio_available() -> bool:
"""Return True if audio libraries can be imported."""
try:
Expand Down Expand Up @@ -296,9 +345,11 @@ def play_beep(frequency: int = 880, duration: float = 0.12, count: int = 1) -> N
duration: Duration of each beep in seconds.
count: Number of beeps to play (with short gap between).
"""
# Synthesize the tone with numpy only (no sounddevice import yet, so the
# macOS TCC prompt is not triggered on the synthesis step).
try:
sd, np = _import_audio()
except (ImportError, OSError):
np = _import_numpy()
except ImportError:
return
try:
gap = 0.06 # seconds between beeps
Expand All @@ -318,6 +369,16 @@ def play_beep(frequency: int = 880, duration: float = 0.12, count: int = 1) -> N
parts.append(np.zeros(samples_per_gap, dtype=np.int16))

audio = np.concatenate(parts)

# On macOS, route the tone through afplay instead of sounddevice.
if not _sounddevice_output_allowed():
_play_int16_via_tempfile(audio, SAMPLE_RATE)
return

try:
sd, _ = _import_audio()
except (ImportError, OSError):
return
sd.play(audio, samplerate=SAMPLE_RATE)
# sd.wait() calls Event.wait() without timeout — hangs forever if the
# audio device stalls. Poll with a 2s ceiling and force-stop.
Expand Down Expand Up @@ -1059,8 +1120,11 @@ def play_audio_file(file_path: str) -> bool:
logger.warning("Audio file not found: %s", file_path)
return False

# Try sounddevice for WAV files
if file_path.endswith(".wav"):
# Skip sounddevice for output where it is not allowed (macOS): PortAudio/
# CoreAudio init triggers a kTCCServiceMediaLibrary permission prompt even
# though playback needs no media-library access. afplay (added to the
# system-player list below) handles all formats natively instead.
if file_path.endswith(".wav") and _sounddevice_output_allowed():
try:
sd, np = _import_audio()
with wave.open(file_path, "rb") as wf:
Expand Down