Add entry tone (Nextel chirp) before queued speech playback

Generate 48kHz WAV tones at startup (chirp and apollo/quindar) using
numpy. The queue consumer plays the selected tone before each speech
item via pw-play. Configurable via TTS_ENTRY_TONE env var: chirp
(default), apollo, none, or path to a custom WAV file.
This commit is contained in:
Ryan Malloy 2026-02-23 14:40:36 -07:00
parent bf0dfa7a5e
commit 25de529bf4
4 changed files with 127 additions and 3 deletions

View File

@ -14,6 +14,7 @@ import sys
import time import time
from dataclasses import dataclass, field from dataclasses import dataclass, field
from enum import IntEnum from enum import IntEnum
from pathlib import Path
from typing import Callable, Coroutine from typing import Callable, Coroutine
from .audio import PlaybackError, play_audio from .audio import PlaybackError, play_audio
@ -50,7 +51,9 @@ class SpeechQueue:
drain pending items. drain pending items.
""" """
def __init__(self, max_depth: int = MAX_QUEUE_DEPTH) -> None: def __init__(
self, max_depth: int = MAX_QUEUE_DEPTH, entry_tone: Path | None = None
) -> None:
self._queue: asyncio.PriorityQueue[_WorkItem] = asyncio.PriorityQueue( self._queue: asyncio.PriorityQueue[_WorkItem] = asyncio.PriorityQueue(
maxsize=max_depth maxsize=max_depth
) )
@ -60,6 +63,7 @@ class SpeechQueue:
self._counter = 0 self._counter = 0
self._max_depth = max_depth self._max_depth = max_depth
self._stopped = False self._stopped = False
self._entry_tone = entry_tone
def _next_id(self) -> str: def _next_id(self) -> str:
self._counter += 1 self._counter += 1
@ -134,6 +138,14 @@ class SpeechQueue:
) )
except Exception: except Exception:
pass # Context may have expired — non-fatal pass # Context may have expired — non-fatal
# Play entry tone (chirp/quindar) before speech
if self._entry_tone:
try:
await play_audio(self._entry_tone, expected_seconds=0.3)
except PlaybackError:
pass # Non-fatal — continue with speech
await play_audio( await play_audio(
item.result.audio_path, item.result.audio_path,
expected_seconds=item.result.duration_seconds, expected_seconds=item.result.duration_seconds,

View File

@ -15,6 +15,7 @@ from .engines.orpheus import OrpheusEngine
from .engines.piper import PiperEngine from .engines.piper import PiperEngine
from .queue import Priority, SpeechQueue from .queue import Priority, SpeechQueue
from .settings import settings from .settings import settings
from .tones import generate_tones, resolve_entry_tone
ENGINE_NAMES = Literal["piper", "kokoro", "orpheus"] ENGINE_NAMES = Literal["piper", "kokoro", "orpheus"]
@ -52,12 +53,17 @@ async def app_lifespan(server: FastMCP):
health = await eng.check_health() health = await eng.check_health()
print(f" {name}: {health['status']}", file=sys.stderr) print(f" {name}: {health['status']}", file=sys.stderr)
queue = SpeechQueue() # Generate entry tones and resolve which one to use
tone_paths = generate_tones(settings.output_dir)
entry_tone = resolve_entry_tone(settings.entry_tone, tone_paths)
queue = SpeechQueue(entry_tone=entry_tone)
queue.start() queue.start()
tone_label = settings.entry_tone if entry_tone else "none"
print( print(
f"TTS MCP server ready on {settings.host}:{settings.port} " f"TTS MCP server ready on {settings.host}:{settings.port} "
f"with {len(engines)} engines", f"with {len(engines)} engines, entry_tone={tone_label}",
file=sys.stderr, file=sys.stderr,
) )

View File

@ -29,6 +29,9 @@ class Settings(BaseSettings):
# Audio output (empty = system temp dir) # Audio output (empty = system temp dir)
audio_dir: str = "" audio_dir: str = ""
# Entry tone before speech: "chirp", "apollo", "none", or path to custom WAV
entry_tone: str = "chirp"
@property @property
def blacklisted_voices(self) -> set[str]: def blacklisted_voices(self) -> set[str]:
return {v.strip().lower() for v in self.voice_blacklist.split(",") if v.strip()} return {v.strip().lower() for v in self.voice_blacklist.split(",") if v.strip()}

103
src/tts_mcp/tones.py Normal file
View File

@ -0,0 +1,103 @@
"""Entry tone generator — short alert tones played before speech.
Generates WAV files programmatically using numpy. Tones are written once
at startup and reused for every queued playback.
"""
import sys
import wave
from pathlib import Path
import numpy as np
SAMPLE_RATE = 48000
def _sine_segment(freq_hz: float, duration_ms: float, fade_ms: float = 1.0) -> np.ndarray:
"""Generate a sine wave segment with fade-in/fade-out to avoid clicks."""
n_samples = int(SAMPLE_RATE * duration_ms / 1000)
t = np.arange(n_samples, dtype=np.float32) / SAMPLE_RATE
tone = np.sin(2 * np.pi * freq_hz * t)
# Apply fade envelope
fade_samples = int(SAMPLE_RATE * fade_ms / 1000)
if fade_samples > 0 and fade_samples * 2 < n_samples:
tone[:fade_samples] *= np.linspace(0, 1, fade_samples)
tone[-fade_samples:] *= np.linspace(1, 0, fade_samples)
return tone
def _silence(duration_ms: float) -> np.ndarray:
return np.zeros(int(SAMPLE_RATE * duration_ms / 1000), dtype=np.float32)
def _build_chirp() -> np.ndarray:
"""Nextel TPT (Talk Permit Tone): 1800 Hz, 24/24/24/24/48 ms pattern."""
segments = [
_sine_segment(1800, 24),
_silence(24),
_sine_segment(1800, 24),
_silence(24),
_sine_segment(1800, 48),
]
return np.concatenate(segments)
def _build_apollo() -> np.ndarray:
"""Quindar intro tone: 2525 Hz, 250 ms with 5 ms fade in/out."""
return _sine_segment(2525, 250, fade_ms=5.0)
def _write_tone(samples: np.ndarray, path: Path) -> Path:
"""Write float32 samples to 16-bit PCM WAV."""
peak = max(abs(samples.max()), abs(samples.min()), 1e-8)
pcm = (samples / peak * 32767 * 0.7).astype(np.int16) # -3 dB headroom
with wave.open(str(path), "wb") as wf:
wf.setnchannels(1)
wf.setsampwidth(2)
wf.setframerate(SAMPLE_RATE)
wf.writeframes(pcm.tobytes())
return path
def generate_tones(output_dir: Path) -> dict[str, Path]:
"""Generate entry tone WAV files. Returns {name: path} mapping."""
output_dir.mkdir(parents=True, exist_ok=True)
tones = {}
for name, builder in [("chirp", _build_chirp), ("apollo", _build_apollo)]:
path = output_dir / f"_tone-{name}.wav"
if not path.exists():
_write_tone(builder(), path)
print(f" Generated entry tone: {path.name}", file=sys.stderr)
tones[name] = path
return tones
def resolve_entry_tone(setting: str, tone_paths: dict[str, Path]) -> Path | None:
"""Resolve the entry_tone setting to a WAV file path or None.
Args:
setting: "chirp", "apollo", "none", or a file path.
tone_paths: Mapping from generate_tones().
"""
if setting == "none":
return None
if setting in tone_paths:
return tone_paths[setting]
# Treat as a custom WAV path
custom = Path(setting)
if custom.exists():
return custom
print(
f" Warning: entry_tone '{setting}' not found, falling back to chirp",
file=sys.stderr,
)
return tone_paths.get("chirp")