from __future__ import annotations import hashlib import os import subprocess import wave from pathlib import Path from time_machine.domain.models import AudioResult, Persona, VoiceProfile class SapiTTSAdapter: """Fast local Windows SAPI TTS adapter for low-latency Walk testing.""" def __init__( self, output_dir: Path | str = "data/audio", script_path: Path | str | None = None, voice_id: str = "windows-sapi", ) -> None: self.output_dir = Path(output_dir) self.voice_id = voice_id self.script_path = ( Path(script_path) if script_path is not None else Path(__file__).resolve().parents[4] / "scripts" / "sapi_tts.ps1" ) def prepare_voice(self, persona: Persona) -> VoiceProfile: return VoiceProfile( voice_id=self.voice_id, description=( f"Windows SAPI voice for {persona.name}: {persona.speaking_style}. " "Used for fast local Walk testing." ), pace="fast", emotion="curious", accent_hint=persona.local_identifier, ) def synthesize( self, text: str, voice_profile: VoiceProfile, prosody_hint: str | None = None, ) -> AudioResult: if os.name != "nt": raise RuntimeError("SapiTTSAdapter requires Windows.") if not self.script_path.exists(): raise RuntimeError(f"Missing SAPI synthesis script: {self.script_path}") self.output_dir.mkdir(parents=True, exist_ok=True) cache_key = self._cache_key(text, voice_profile, prosody_hint) audio_path = self.output_dir / f"sapi-{cache_key}.wav" if audio_path.exists(): return AudioResult( path=str(audio_path), mime_type="audio/wav", duration_seconds=_wav_duration(audio_path), description=f"Cached Windows SAPI synthesis using {voice_profile.voice_id}.", ) text_path = self.output_dir / f"sapi-{cache_key}.txt" text_path.write_text(text, encoding="utf-8") rate = {"slow": -2, "medium": 0, "fast": 2}[voice_profile.pace] completed = subprocess.run( [ "powershell.exe", "-NoProfile", "-ExecutionPolicy", "Bypass", "-File", str(self.script_path), "-TextPath", str(text_path), "-OutputPath", str(audio_path), "-Rate", str(rate), ], check=False, capture_output=True, text=True, timeout=int(os.getenv("TIME_MACHINE_SAPI_TIMEOUT_SECONDS", "30")), ) if completed.returncode != 0: raise RuntimeError( "Windows SAPI synthesis failed: " f"{completed.stderr.strip() or completed.stdout.strip()}" ) return AudioResult( path=str(audio_path), mime_type="audio/wav", duration_seconds=_wav_duration(audio_path), description=f"Windows SAPI synthesis using {voice_profile.voice_id}.", ) def warm_up(self, voice_profile: VoiceProfile | None = None) -> None: return None def _cache_key( self, text: str, voice_profile: VoiceProfile, prosody_hint: str | None, ) -> str: payload = "\n".join( [ self.voice_id, voice_profile.pace, prosody_hint or "", text, ] ) return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:16] def _wav_duration(path: Path) -> float | None: try: with wave.open(str(path), "rb") as handle: frames = handle.getnframes() rate = handle.getframerate() if rate <= 0: return None return round(frames / rate, 3) except wave.Error: return None