src/time_machine/adapters/tts/sapi.py

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