Add exit tone (roger beep) and standby tone after speech

Queue-aware exit tones: descending roger beep when queue empties
("over"), ascending standby blip when more items are queued ("standby,
more coming"). Includes quindar-out (2475 Hz) as Apollo-themed
alternative. Configurable via TTS_EXIT_TONE env var.

Also adds CLAUDE.md documenting the full tone system.
This commit is contained in:
Ryan Malloy 2026-02-23 14:44:27 -07:00
parent 25de529bf4
commit 6a81e0760a
5 changed files with 145 additions and 19 deletions

61
CLAUDE.md Normal file
View File

@ -0,0 +1,61 @@
# TTS MCP Server
Multi-engine text-to-speech server exposed via FastMCP 3.0 Streamable HTTP. Engines: Kokoro (ONNX), Piper (Wyoming/Docker), Orpheus (llama-server + SNAC).
## Build & Run
```bash
make up # build + start (docker compose)
make logs # follow logs
make restart # restart containers
make status # show running containers + health
```
## Entry & Exit Tones (Beep System)
Queued speech playback (`speak()`) is bookended by short alert tones. `generate_audio()` is unaffected (file-only, no playback).
### Tone Positions
| Position | When | Purpose |
|----------|------|---------|
| **Entry tone** | Before speech starts | "Incoming transmission" alert |
| **Exit tone** | After speech, queue empty | "Over and out" — channel clear |
| **Standby tone** | After speech, more queued | "Standby" — more messages coming |
### Available Tones
| Name | Frequency | Duration | Inspired by |
|------|-----------|----------|-------------|
| `chirp` | 1800 Hz | ~144 ms | Nextel iDEN Talk Permit Tone (TPT) — the 24/24/24/24/48 ms on/off pattern |
| `apollo` | 2525 Hz | 250 ms | NASA quindar intro (key-up) tone used during Apollo missions |
| `roger` | 1400-1000 Hz | ~100 ms | Classic CB radio descending two-tone roger beep |
| `quindar-out` | 2475 Hz | 250 ms | NASA quindar unkey tone (distinct frequency from intro) |
| `standby` | 1000-1400 Hz | ~60 ms | Ascending blip — inverse of roger, signals "more coming" |
### Configuration
```env
TTS_ENTRY_TONE=chirp # before speech (chirp, apollo, none, or /path/to/custom.wav)
TTS_EXIT_TONE=roger # after speech, queue empty (roger, quindar-out, none, or path)
```
The standby tone is always the built-in ascending blip. It plays instead of the exit tone when more items are queued.
Tones are generated programmatically at startup (48kHz, 16-bit PCM, -3 dB headroom) in `tones.py` using numpy. No bundled audio assets.
## Architecture
- `server.py` — FastMCP lifespan, tool definitions, engine setup
- `queue.py` — Producer-consumer speech queue with priority tiers
- `tones.py` — Tone WAV generator (entry/exit/standby)
- `audio.py` — WAV writing and `pw-play` async wrapper
- `settings.py` — Pydantic settings from env vars (prefix: `TTS_`)
- `engines/` — TTSEngine implementations (kokoro, piper, orpheus)
## Key Design Decisions
- Speech queue is serialized (one playback at a time) but synthesis is parallel
- Tones are non-fatal: if `pw-play` fails on a tone, speech still plays
- Orpheus uses llama-server (not Ollama) for 15x throughput via continuous batching
- SNAC decoder is lazy-loaded on first Orpheus call to reduce idle memory

View File

@ -52,7 +52,11 @@ class SpeechQueue:
""" """
def __init__( def __init__(
self, max_depth: int = MAX_QUEUE_DEPTH, entry_tone: Path | None = None self,
max_depth: int = MAX_QUEUE_DEPTH,
entry_tone: Path | None = None,
exit_tone: Path | None = None,
standby_tone: Path | None = None,
) -> None: ) -> None:
self._queue: asyncio.PriorityQueue[_WorkItem] = asyncio.PriorityQueue( self._queue: asyncio.PriorityQueue[_WorkItem] = asyncio.PriorityQueue(
maxsize=max_depth maxsize=max_depth
@ -64,6 +68,8 @@ class SpeechQueue:
self._max_depth = max_depth self._max_depth = max_depth
self._stopped = False self._stopped = False
self._entry_tone = entry_tone self._entry_tone = entry_tone
self._exit_tone = exit_tone
self._standby_tone = standby_tone
def _next_id(self) -> str: def _next_id(self) -> str:
self._counter += 1 self._counter += 1
@ -150,6 +156,18 @@ class SpeechQueue:
item.result.audio_path, item.result.audio_path,
expected_seconds=item.result.duration_seconds, expected_seconds=item.result.duration_seconds,
) )
# Play exit tone — "standby" if more queued, "roger" if done
exit = (
self._standby_tone if self._queue.qsize() > 0
else self._exit_tone
)
if exit:
try:
await play_audio(exit, expected_seconds=0.3)
except PlaybackError:
pass # Non-fatal
if not item.future.done(): if not item.future.done():
item.future.set_result({ item.future.set_result({
"played": True, "played": True,

View File

@ -15,7 +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 from .tones import generate_tones, resolve_tone
ENGINE_NAMES = Literal["piper", "kokoro", "orpheus"] ENGINE_NAMES = Literal["piper", "kokoro", "orpheus"]
@ -53,17 +53,20 @@ 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)
# Generate entry tones and resolve which one to use # Generate tones and resolve which ones to use
tone_paths = generate_tones(settings.output_dir) tone_paths = generate_tones(settings.output_dir)
entry_tone = resolve_entry_tone(settings.entry_tone, tone_paths) entry_tone = resolve_tone(settings.entry_tone, tone_paths, "entry_tone", default="chirp")
exit_tone = resolve_tone(settings.exit_tone, tone_paths, "exit_tone", default="roger")
standby_tone = tone_paths.get("standby")
queue = SpeechQueue(entry_tone=entry_tone) queue = SpeechQueue(entry_tone=entry_tone, exit_tone=exit_tone, standby_tone=standby_tone)
queue.start() queue.start()
tone_label = settings.entry_tone if entry_tone else "none" entry_label = settings.entry_tone if entry_tone else "none"
exit_label = settings.exit_tone if exit_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, entry_tone={tone_label}", f"with {len(engines)} engines, tones={entry_label}/{exit_label}",
file=sys.stderr, file=sys.stderr,
) )

View File

@ -32,6 +32,10 @@ class Settings(BaseSettings):
# Entry tone before speech: "chirp", "apollo", "none", or path to custom WAV # Entry tone before speech: "chirp", "apollo", "none", or path to custom WAV
entry_tone: str = "chirp" entry_tone: str = "chirp"
# Exit tone after speech: "roger", "quindar-out", "none", or path to custom WAV
# When more items are queued, plays "standby" (ascending blip) instead
exit_tone: str = "roger"
@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()}

View File

@ -1,4 +1,4 @@
"""Entry tone generator — short alert tones played before speech. """Tone generator — short alert tones played before/after speech.
Generates WAV files programmatically using numpy. Tones are written once Generates WAV files programmatically using numpy. Tones are written once
at startup and reused for every queued playback. at startup and reused for every queued playback.
@ -49,6 +49,29 @@ def _build_apollo() -> np.ndarray:
return _sine_segment(2525, 250, fade_ms=5.0) return _sine_segment(2525, 250, fade_ms=5.0)
def _build_roger() -> np.ndarray:
"""Classic descending two-tone roger beep: 1400 Hz -> 1000 Hz."""
segments = [
_sine_segment(1400, 40),
_sine_segment(1000, 60),
]
return np.concatenate(segments)
def _build_quindar_out() -> np.ndarray:
"""Quindar unkey tone: 2475 Hz, 250 ms with 5 ms fade in/out."""
return _sine_segment(2475, 250, fade_ms=5.0)
def _build_standby() -> np.ndarray:
"""Quick ascending blip — "more coming, standby"."""
segments = [
_sine_segment(1000, 30),
_sine_segment(1400, 30),
]
return np.concatenate(segments)
def _write_tone(samples: np.ndarray, path: Path) -> Path: def _write_tone(samples: np.ndarray, path: Path) -> Path:
"""Write float32 samples to 16-bit PCM WAV.""" """Write float32 samples to 16-bit PCM WAV."""
peak = max(abs(samples.max()), abs(samples.min()), 1e-8) peak = max(abs(samples.max()), abs(samples.min()), 1e-8)
@ -63,27 +86,40 @@ def _write_tone(samples: np.ndarray, path: Path) -> Path:
return path return path
_TONE_BUILDERS = {
"chirp": _build_chirp,
"apollo": _build_apollo,
"roger": _build_roger,
"quindar-out": _build_quindar_out,
"standby": _build_standby,
}
def generate_tones(output_dir: Path) -> dict[str, Path]: def generate_tones(output_dir: Path) -> dict[str, Path]:
"""Generate entry tone WAV files. Returns {name: path} mapping.""" """Generate all tone WAV files. Returns {name: path} mapping."""
output_dir.mkdir(parents=True, exist_ok=True) output_dir.mkdir(parents=True, exist_ok=True)
tones = {} tones = {}
for name, builder in [("chirp", _build_chirp), ("apollo", _build_apollo)]: for name, builder in _TONE_BUILDERS.items():
path = output_dir / f"_tone-{name}.wav" path = output_dir / f"_tone-{name}.wav"
if not path.exists(): if not path.exists():
_write_tone(builder(), path) _write_tone(builder(), path)
print(f" Generated entry tone: {path.name}", file=sys.stderr) print(f" Generated tone: {path.name}", file=sys.stderr)
tones[name] = path tones[name] = path
return tones return tones
def resolve_entry_tone(setting: str, tone_paths: dict[str, Path]) -> Path | None: def resolve_tone(
"""Resolve the entry_tone setting to a WAV file path or None. setting: str, tone_paths: dict[str, Path], label: str, default: str | None = None
) -> Path | None:
"""Resolve a tone setting to a WAV file path or None.
Args: Args:
setting: "chirp", "apollo", "none", or a file path. setting: Built-in tone name, "none", or a file path.
tone_paths: Mapping from generate_tones(). tone_paths: Mapping from generate_tones().
label: Setting name for warning messages (e.g. "entry_tone").
default: Fallback tone name if setting not found. None = no fallback.
""" """
if setting == "none": if setting == "none":
return None return None
@ -96,8 +132,12 @@ def resolve_entry_tone(setting: str, tone_paths: dict[str, Path]) -> Path | None
if custom.exists(): if custom.exists():
return custom return custom
print( fallback = tone_paths.get(default) if default else None
f" Warning: entry_tone '{setting}' not found, falling back to chirp", if fallback:
file=sys.stderr, print(
) f" Warning: {label} '{setting}' not found, falling back to {default}",
return tone_paths.get("chirp") file=sys.stderr,
)
else:
print(f" Warning: {label} '{setting}' not found, disabled", file=sys.stderr)
return fallback