From 70511916b9d76c2a6db271de0748331a40a78c0f Mon Sep 17 00:00:00 2001 From: Ryan Malloy Date: Tue, 3 Mar 2026 21:38:56 -0700 Subject: [PATCH] Add media ducking: fade external audio during TTS playback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Duck all PulseAudio sink-inputs (Firefox, Spotify, etc.) before speech, crossfade the entry tone over the fading media, then restore volumes after the exit tone. Uses pactl subprocesses for reliable PipeWire compatibility — pulsectl-asyncio's native protocol writes are silently dropped by PipeWire's PA compat layer. New module media_duck.py with MediaDucker class. Unduck is consolidated into the consumer's finally block to guarantee restoration on all exit paths (normal, error, cancel, shutdown). --- CLAUDE.md | 48 +++++++++++ Dockerfile | 2 + docker-compose.yml | 4 + src/tts_mcp/media_duck.py | 177 ++++++++++++++++++++++++++++++++++++++ src/tts_mcp/queue.py | 20 ++++- src/tts_mcp/server.py | 69 ++++++++++++--- src/tts_mcp/settings.py | 5 ++ 7 files changed, 309 insertions(+), 16 deletions(-) create mode 100644 src/tts_mcp/media_duck.py diff --git a/CLAUDE.md b/CLAUDE.md index 264aa08..cb272f6 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -82,11 +82,59 @@ Voices are interleaved for perceptual diversity: American female → British mal - `voice_identity.py` — Pool filtering, interleaving, round-robin assignment, JSON persistence +## Media Ducking + +When `speak()` is called, external audio streams (Firefox, Spotify, etc.) are automatically faded down via PulseAudio before speech begins, then faded back up after the exit tone. This creates a radio-broadcast-interruption effect where the entry tone crossfades over the fading media. + +### Audio Timeline + +``` +0ms ─ Media starts fading (vol 100%) +150ms ─ Entry tone starts! Media at ~70% ← crossfade overlap +294ms ─ Entry tone ends, media at ~40% +500ms ─ Media at 0% (ducked) + ─ [synthesis + playback — media stays silent] + ─ Exit tone plays (roger/standby) + ─ Media fades back in over 1000ms +``` + +The duck fires in `speak()` (before synthesis), the unduck fires in the consumer (after exit/cancel tone). Media stays ducked across multiple queued items — only unducks when the queue is empty or on cancel/shutdown. + +### Configuration + +```env +TTS_DUCK_MEDIA=true # Enable/disable (on by default) +TTS_DUCK_FADE_OUT_MS=500 # Fade-out duration (ms) +TTS_DUCK_FADE_IN_MS=1000 # Fade-in duration (slower = natural) +``` + +### Docker Requirements + +The PulseAudio compatibility socket must be mounted in the container: + +```yaml +volumes: + - /run/user/1000/pulse:/run/user/1000/pulse +``` + +If the socket is missing or pactl fails, ducking silently no-ops — TTS still works normally. + +**Why pactl, not pulsectl-asyncio?** PipeWire's PulseAudio compat layer silently drops `sink_input_volume_set` operations from pulsectl's native protocol connection, while `pactl` (using libpulse C library) works reliably. The `pulseaudio-utils` package is installed in the container for this reason. + +### Non-fatal Design + +All pactl errors are caught and logged. If PulseAudio is unavailable (no socket, wrong permissions, pactl not installed), duck/unduck become no-ops. This ensures TTS never breaks due to ducking failures. + +### Files + +- `media_duck.py` — `MediaDucker` class (async pactl subprocess volume control with stepped fades) + ## Architecture - `server.py` — FastMCP lifespan, tool definitions, engine setup - `queue.py` — Producer-consumer speech queue with priority tiers and outcome tracking - `tones.py` — Tone WAV generator (entry/exit/standby) +- `media_duck.py` — Async PulseAudio volume control for media ducking - `audio.py` — WAV writing and `pw-play` async wrapper - `settings.py` — Pydantic settings from env vars (prefix: `TTS_`) - `engines/` — TTSEngine implementations (kokoro, piper, orpheus) diff --git a/Dockerfile b/Dockerfile index ee1cc14..358424c 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,8 +1,10 @@ FROM ghcr.io/astral-sh/uv:python3.13-bookworm-slim # PipeWire client tools — pw-play connects to the host's PipeWire socket +# pulseaudio-utils provides pactl for media ducking (volume control via PulseAudio compat) RUN apt-get update && apt-get install -y --no-install-recommends \ pipewire-bin \ + pulseaudio-utils \ && rm -rf /var/lib/apt/lists/* WORKDIR /app diff --git a/docker-compose.yml b/docker-compose.yml index eb08952..9b31f03 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -13,6 +13,8 @@ services: TTS_ORPHEUS_URL: http://llama-server:8081 # PipeWire client config XDG_RUNTIME_DIR: /run/user/1000 + # PulseAudio socket for media ducking (bypass XDG_RUNTIME_DIR ownership check) + PULSE_SERVER: unix:/run/user/1000/pulse/native # Force unbuffered Python output so stderr shows up in docker logs immediately PYTHONUNBUFFERED: "1" volumes: @@ -24,6 +26,8 @@ services: - tts-data:/data # PipeWire socket for audio playback through host speakers - /run/user/1000/pipewire-0:/run/user/1000/pipewire-0 + # PulseAudio compat socket for media ducking (volume control) + - /run/user/1000/pulse:/run/user/1000/pulse depends_on: llama-server: condition: service_healthy diff --git a/src/tts_mcp/media_duck.py b/src/tts_mcp/media_duck.py new file mode 100644 index 0000000..0124751 --- /dev/null +++ b/src/tts_mcp/media_duck.py @@ -0,0 +1,177 @@ +"""Media ducking — fade external audio during TTS playback. + +Uses pactl (PulseAudio CLI) to control sink-input volumes through PipeWire's +PulseAudio compatibility layer. When duck() is called, all active sink-inputs +(Firefox, Spotify, etc.) have their volumes stepped down over a configurable +fade period. unduck() restores them. + +We use pactl as a subprocess rather than pulsectl-asyncio because PipeWire's +PA compat layer silently drops volume-set operations from pulsectl's native +protocol connection, while pactl (using libpulse) works reliably. + +If pactl is not installed or the PulseAudio socket is unavailable, all +operations silently no-op — TTS still works, just without ducking. +""" + +import asyncio +import re +import shutil +import sys +from dataclasses import dataclass, field + +_HAS_PACTL = shutil.which("pactl") is not None + + +@dataclass +class _SavedStream: + """Original volume state for a single sink-input.""" + + index: int + app_name: str + volume_pct: int # percentage (0-100+), per PulseAudio's volume scale + + +@dataclass +class MediaDucker: + """Async PulseAudio volume controller for media ducking. + + Fade-out steps down all external sink-input volumes over `fade_out_ms`. + Fade-in restores saved volumes over `fade_in_ms`. Both use ~10 steps + for smooth animation. + + All errors are caught and logged — never blocks TTS playback. + """ + + fade_out_ms: int = 500 + fade_in_ms: int = 1000 + _is_ducked: bool = field(default=False, init=False) + _saved_streams: list[_SavedStream] = field(default_factory=list, init=False) + _lock: asyncio.Lock = field(default_factory=asyncio.Lock, init=False) + + async def duck(self) -> None: + """Fade down all external audio streams. No-op if already ducked.""" + if not _HAS_PACTL: + return + async with self._lock: + if self._is_ducked: + return + try: + await self._do_duck() + self._is_ducked = True + except Exception as e: + print(f" Media duck failed (non-fatal): {e}", file=sys.stderr) + # Rollback partially-ducked streams to avoid the "volume ratchet" + try: + await self._do_unduck() + except Exception: + pass + self._saved_streams.clear() + + async def unduck(self) -> None: + """Restore all audio streams to original volumes. No-op if not ducked.""" + if not _HAS_PACTL: + return + async with self._lock: + if not self._is_ducked: + return + try: + await self._do_unduck() + except Exception as e: + print(f" Media unduck failed (non-fatal): {e}", file=sys.stderr) + finally: + self._is_ducked = False + self._saved_streams.clear() + + async def _pactl_set_volume(self, sink_input_idx: int, volume_pct: int) -> None: + """Set a sink-input's volume via pactl subprocess.""" + proc = await asyncio.create_subprocess_exec( + "pactl", "set-sink-input-volume", str(sink_input_idx), f"{volume_pct}%", + stdout=asyncio.subprocess.DEVNULL, + stderr=asyncio.subprocess.DEVNULL, + ) + await proc.wait() + + async def _list_sink_inputs(self) -> list[_SavedStream]: + """Parse pactl list sink-inputs to get index, app name, and volume.""" + proc = await asyncio.create_subprocess_exec( + "pactl", "list", "sink-inputs", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.DEVNULL, + ) + stdout, _ = await proc.communicate() + if proc.returncode != 0: + return [] + + streams = [] + text = stdout.decode(errors="replace") + + # Split into per-sink-input blocks + blocks = re.split(r"^Sink Input #", text, flags=re.MULTILINE) + for block in blocks[1:]: # skip preamble before first match + # Extract index + idx_match = re.match(r"(\d+)", block) + if not idx_match: + continue + idx = int(idx_match.group(1)) + + # Extract volume percentage (take first channel) + vol_match = re.search(r"Volume:.*?(\d+)%", block) + vol_pct = int(vol_match.group(1)) if vol_match else 100 + + # Extract app name + app_match = re.search(r'application\.name\s*=\s*"([^"]*)"', block) + app_name = app_match.group(1) if app_match else "unknown" + + streams.append(_SavedStream(index=idx, app_name=app_name, volume_pct=vol_pct)) + + return streams + + async def _do_duck(self) -> None: + """Internal: enumerate sink-inputs and step volumes to zero.""" + streams = await self._list_sink_inputs() + if not streams: + print(" Media duck: no sink-inputs found", file=sys.stderr) + return + + self._saved_streams = streams + apps = [s.app_name for s in streams] + print( + f" Media duck: fading {len(streams)} streams ({', '.join(apps)})", + file=sys.stderr, + ) + + # Step down in ~10 steps + n_steps = 10 + step_delay = self.fade_out_ms / 1000.0 / n_steps + + for step in range(1, n_steps + 1): + fraction = 1.0 - (step / n_steps) # 0.9 → 0.0 + tasks = [] + for s in self._saved_streams: + target_pct = max(int(s.volume_pct * fraction), 0) + tasks.append(self._pactl_set_volume(s.index, target_pct)) + await asyncio.gather(*tasks, return_exceptions=True) + await asyncio.sleep(step_delay) + + async def _do_unduck(self) -> None: + """Internal: restore saved volumes with fade-in.""" + if not self._saved_streams: + return + + print( + f" Media unduck: restoring {len(self._saved_streams)} streams", + file=sys.stderr, + ) + + # Step up in ~10 steps + n_steps = 10 + step_delay = self.fade_in_ms / 1000.0 / n_steps + + for step in range(1, n_steps + 1): + fraction = step / n_steps # 0.1 → 1.0 + tasks = [] + for s in self._saved_streams: + target_pct = int(s.volume_pct * fraction) + tasks.append(self._pactl_set_volume(s.index, target_pct)) + await asyncio.gather(*tasks, return_exceptions=True) + await asyncio.sleep(step_delay) diff --git a/src/tts_mcp/queue.py b/src/tts_mcp/queue.py index 4195659..41a0db2 100644 --- a/src/tts_mcp/queue.py +++ b/src/tts_mcp/queue.py @@ -20,6 +20,7 @@ from pathlib import Path from .audio import PlaybackError, play_audio from .engines.base import TTSResult +from .media_duck import MediaDucker class Priority(IntEnum): @@ -58,6 +59,7 @@ class SpeechQueue: standby_tone: Path | None = None, cancel_tone: Path | None = None, shutdown_timeout: float = 30.0, + ducker: MediaDucker | None = None, ) -> None: self._queue: asyncio.PriorityQueue[_WorkItem] = asyncio.PriorityQueue( maxsize=max_depth @@ -72,6 +74,7 @@ class SpeechQueue: self._standby_tone = standby_tone self._cancel_tone = cancel_tone self._shutdown_timeout = shutdown_timeout + self._ducker = ducker self._outcomes: OrderedDict[str, dict] = OrderedDict() self._futures: dict[str, asyncio.Future] = {} @@ -178,6 +181,12 @@ class SpeechQueue: self._force_cancel_consumer() raise finally: + # Always restore media volumes on shutdown + if self._ducker: + try: + await self._ducker.unduck() + except Exception: + pass # Non-fatal self._drain_remaining() def _resolve_outcome(self, item: _WorkItem, outcome: dict) -> None: @@ -217,8 +226,8 @@ class SpeechQueue: # suppress_exit_tone skips tones between chunks of the same message if not item.suppress_exit_tone: exit_t = ( - self._standby_tone if self._queue.qsize() > 0 - else self._exit_tone + self._exit_tone if self._queue.qsize() == 0 + else self._standby_tone ) if exit_t: try: @@ -280,6 +289,13 @@ class SpeechQueue: finally: self._current = None self._queue.task_done() + # Unduck media when queue is empty — covers ALL exit paths + # (normal completion, PlaybackError, cancel, generic exception) + if self._queue.qsize() == 0 and self._ducker and not self._stopped: + try: + await self._ducker.unduck() + except Exception: + pass # Non-fatal async def enqueue( self, diff --git a/src/tts_mcp/server.py b/src/tts_mcp/server.py index 7580928..5bfb454 100644 --- a/src/tts_mcp/server.py +++ b/src/tts_mcp/server.py @@ -17,6 +17,7 @@ from .engines.base import TTSEngine from .engines.kokoro import KokoroEngine from .engines.orpheus import OrpheusEngine from .engines.piper import PiperEngine +from .media_duck import MediaDucker from .queue import Priority, SpeechQueue from .settings import settings from .tones import generate_tones, resolve_tone @@ -65,20 +66,31 @@ async def app_lifespan(server: FastMCP): cancel_tone = resolve_tone(settings.cancel_tone, tone_paths, "cancel_tone", default="scratch") standby_tone = tone_paths.get("standby") + # --- Media ducker --- + ducker: MediaDucker | None = None + if settings.duck_media: + ducker = MediaDucker( + fade_out_ms=settings.duck_fade_out_ms, + fade_in_ms=settings.duck_fade_in_ms, + ) + queue = SpeechQueue( exit_tone=exit_tone, standby_tone=standby_tone, cancel_tone=cancel_tone, shutdown_timeout=settings.shutdown_timeout, + ducker=ducker, ) queue.start() entry_label = settings.entry_tone if entry_tone else "none" exit_label = settings.exit_tone if exit_tone else "none" cancel_label = settings.cancel_tone if cancel_tone else "none" + duck_label = "on" if ducker else "off" print( f"TTS MCP server ready on {settings.host}:{settings.port} " - f"with {len(engines)} engines, tones={entry_label}/{exit_label}/{cancel_label}", + f"with {len(engines)} engines, tones={entry_label}/{exit_label}/{cancel_label}, " + f"duck={duck_label}", file=sys.stderr, ) @@ -101,6 +113,7 @@ async def app_lifespan(server: FastMCP): "queue": queue, "voice_cache": voice_cache, "entry_tone": entry_tone, + "ducker": ducker, } finally: print("TTS MCP server shutting down", file=sys.stderr) @@ -132,10 +145,18 @@ mcp = FastMCP( ) -def _get_state(ctx: Context) -> tuple[dict[str, TTSEngine], SpeechQueue, VoiceIdentityCache, Path | None]: - """Extract engines, queue, voice cache, and entry tone from lifespan context.""" +def _get_state( + ctx: Context, +) -> tuple[dict[str, TTSEngine], SpeechQueue, VoiceIdentityCache, Path | None, MediaDucker | None]: + """Extract engines, queue, voice cache, entry tone, and ducker from lifespan context.""" state = ctx.lifespan_context - return state["engines"], state["queue"], state["voice_cache"], state["entry_tone"] + return ( + state["engines"], + state["queue"], + state["voice_cache"], + state["entry_tone"], + state["ducker"], + ) async def _play_tone(tone_path: Path) -> None: @@ -221,6 +242,7 @@ async def _speak_single( entry_tone: Path | None, priority: Priority, ctx: Context, + ducker: MediaDucker | None = None, ) -> dict: """Single-shot speak path — synthesize full text, then enqueue. @@ -228,7 +250,14 @@ async def _speak_single( no sentence boundaries. """ try: - if entry_tone: + # Duck external media with crossfade into entry tone + if ducker: + duck_task = asyncio.create_task(ducker.duck()) + await asyncio.sleep(0.15) # Let fade start — media dips to ~70% + if entry_tone: + await _play_tone(entry_tone) # Tone crossfades over fading media + await duck_task # Ensure duck completes + elif entry_tone: await _play_tone(entry_tone) await ctx.report_progress(progress=5, total=100, message="Entry tone") @@ -273,6 +302,7 @@ async def _speak_chunked( entry_tone: Path | None, priority: Priority, ctx: Context, + ducker: MediaDucker | None = None, ) -> dict: """Chunked speak path — pipeline synthesis with playback. @@ -286,7 +316,14 @@ async def _speak_chunked( first_enqueue_time: float | None = None try: - if entry_tone: + # Duck external media with crossfade into entry tone + if ducker: + duck_task = asyncio.create_task(ducker.duck()) + await asyncio.sleep(0.15) # Let fade start — media dips to ~70% + if entry_tone: + await _play_tone(entry_tone) # Tone crossfades over fading media + await duck_task # Ensure duck completes + elif entry_tone: await _play_tone(entry_tone) await ctx.report_progress(progress=5, total=100, message="Entry tone") @@ -431,7 +468,7 @@ async def speak( urgent: If True, this message jumps ahead of normal-priority items. project: Project name for voice identity (auto-detected from MCP roots if omitted). """ - engines, queue, voice_cache, entry_tone = _get_state(ctx) + engines, queue, voice_cache, entry_tone, ducker = _get_state(ctx) if engine not in engines: return {"error": f"Unknown engine: {engine}. Available: {list(engines.keys())}"} @@ -445,8 +482,12 @@ async def speak( chunks = split_text(text) if len(chunks) <= 1: - return await _speak_single(eng, engine, text, voice, queue, entry_tone, priority, ctx) - return await _speak_chunked(eng, engine, chunks, voice, queue, entry_tone, priority, ctx) + return await _speak_single( + eng, engine, text, voice, queue, entry_tone, priority, ctx, ducker, + ) + return await _speak_chunked( + eng, engine, chunks, voice, queue, entry_tone, priority, ctx, ducker, + ) @mcp.tool(annotations=ToolAnnotations(readOnlyHint=True)) @@ -462,7 +503,7 @@ async def speech_status( Args: speech_id: The speech_id returned by speak(). """ - _, queue, _, _ = _get_state(ctx) + _, queue, _, _, _ = _get_state(ctx) return queue.get_status(speech_id) @@ -483,7 +524,7 @@ async def cancel_speech( Args: speech_id: The speech_id returned by speak(). """ - _, queue, _, _ = _get_state(ctx) + _, queue, _, _, _ = _get_state(ctx) return queue.cancel(speech_id) @@ -510,7 +551,7 @@ async def generate_audio( voice: Voice name (use list_voices to see options). None = auto-assigned by project. project: Project name for voice identity (auto-detected from MCP roots if omitted). """ - engines, _, voice_cache, _ = _get_state(ctx) + engines, _, voice_cache, _, _ = _get_state(ctx) if engine not in engines: return {"error": f"Unknown engine: {engine}. Available: {list(engines.keys())}"} @@ -545,7 +586,7 @@ async def list_voices( Args: engine: Which engine to list voices for. """ - engines, _, _, _ = _get_state(ctx) + engines, _, _, _, _ = _get_state(ctx) if engine not in engines: return [] @@ -561,7 +602,7 @@ async def list_engines( Returns engine name, default voice, and health check results. """ - engines, queue, _, _ = _get_state(ctx) + engines, queue, _, _, _ = _get_state(ctx) results = [] for name, eng in engines.items(): diff --git a/src/tts_mcp/settings.py b/src/tts_mcp/settings.py index eb0092e..3a46141 100644 --- a/src/tts_mcp/settings.py +++ b/src/tts_mcp/settings.py @@ -49,6 +49,11 @@ class Settings(BaseSettings): # Cancel tone: "scratch", "reverse-roger", "none", or path to custom WAV cancel_tone: str = "scratch" + # Media ducking + duck_media: bool = True + duck_fade_out_ms: int = 500 + duck_fade_in_ms: int = 1000 + # Graceful shutdown: max seconds to wait for current speech to finish shutdown_timeout: float = 30.0