diff --git a/src/tts_mcp/queue.py b/src/tts_mcp/queue.py index 8bd5674..7631d39 100644 --- a/src/tts_mcp/queue.py +++ b/src/tts_mcp/queue.py @@ -14,6 +14,7 @@ import sys import time from dataclasses import dataclass, field from enum import IntEnum +from pathlib import Path from typing import Callable, Coroutine from .audio import PlaybackError, play_audio @@ -50,7 +51,9 @@ class SpeechQueue: 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( maxsize=max_depth ) @@ -60,6 +63,7 @@ class SpeechQueue: self._counter = 0 self._max_depth = max_depth self._stopped = False + self._entry_tone = entry_tone def _next_id(self) -> str: self._counter += 1 @@ -134,6 +138,14 @@ class SpeechQueue: ) except Exception: 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( item.result.audio_path, expected_seconds=item.result.duration_seconds, diff --git a/src/tts_mcp/server.py b/src/tts_mcp/server.py index 6a31531..e0c7c89 100644 --- a/src/tts_mcp/server.py +++ b/src/tts_mcp/server.py @@ -15,6 +15,7 @@ from .engines.orpheus import OrpheusEngine from .engines.piper import PiperEngine from .queue import Priority, SpeechQueue from .settings import settings +from .tones import generate_tones, resolve_entry_tone ENGINE_NAMES = Literal["piper", "kokoro", "orpheus"] @@ -52,12 +53,17 @@ async def app_lifespan(server: FastMCP): health = await eng.check_health() 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() + tone_label = settings.entry_tone if entry_tone else "none" print( 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, ) diff --git a/src/tts_mcp/settings.py b/src/tts_mcp/settings.py index 24f116e..9659435 100644 --- a/src/tts_mcp/settings.py +++ b/src/tts_mcp/settings.py @@ -29,6 +29,9 @@ class Settings(BaseSettings): # Audio output (empty = system temp dir) audio_dir: str = "" + # Entry tone before speech: "chirp", "apollo", "none", or path to custom WAV + entry_tone: str = "chirp" + @property def blacklisted_voices(self) -> set[str]: return {v.strip().lower() for v in self.voice_blacklist.split(",") if v.strip()} diff --git a/src/tts_mcp/tones.py b/src/tts_mcp/tones.py new file mode 100644 index 0000000..686a3cb --- /dev/null +++ b/src/tts_mcp/tones.py @@ -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")