voice.py

"""
Voice Module - Unified Text-to-Speech Interface

Supports two backends:
- Local: Piper TTS (offline, free, robotic voices)
- Cloud: ElevenLabs (online, paid, natural voices)

Usage:
    from voice import speak, speak_async, set_backend

    # Use local TTS (default)
    speak("Hello world")

    # Switch to ElevenLabs
    set_backend("elevenlabs")
    speak("Hello world")

    # Or specify per-call
    speak("Hello world", backend="elevenlabs")
"""

import os
import sys
import json
import wave
import math
import struct
import tempfile
import threading
from typing import Optional, Literal

# Try to import requests for ElevenLabs
try:
    import requests
    _requests_available = True
except ImportError:
    _requests_available = False

# -----------------------------------------------------------------------------
# Configuration
# -----------------------------------------------------------------------------

Backend = Literal["local", "elevenlabs"]

# Default backend
_current_backend: Backend = "local"

# Mute state
_muted = False

# ElevenLabs settings (can be overridden via config or environment)
ELEVENLABS_API_KEY = os.environ.get("ELEVENLABS_API_KEY", "")
ELEVENLABS_VOICE_ID = os.environ.get("ELEVENLABS_VOICE_ID", "")  # Set your voice ID
ELEVENLABS_MODEL_ID = os.environ.get("ELEVENLABS_MODEL_ID", "eleven_v3")

# Voice settings for ElevenLabs
ELEVENLABS_VOICE_SETTINGS = {
    "stability": 0.5,
    "similarity_boost": 0.75,
    "style": 0.4,
    "use_speaker_boost": True
}

# -----------------------------------------------------------------------------
# Local TTS (Piper) State
# -----------------------------------------------------------------------------

_tts_available = None
_male_voice = None
_female_voice = None
_voices_loaded = False
_voice_lock = threading.Lock()

# Paths
_module_dir = os.path.dirname(os.path.abspath(__file__))
_voices_dir = os.path.join(_module_dir, 'voices')
_beep_file = os.path.join(_module_dir, 'alert_beep.wav')
_startup_beep_file = os.path.join(_module_dir, 'startup_beep.wav')

# Piper voice models
_voice_models = {
    'male': 'en/en_US/danny/low/en_US-danny-low.onnx',
    'female': 'en/en_US/amy/low/en_US-amy-low.onnx',
}

# -----------------------------------------------------------------------------
# Backend Management
# -----------------------------------------------------------------------------

def get_backend() -> Backend:
    """Get the current TTS backend."""
    return _current_backend


def set_backend(backend: Backend):
    """Set the TTS backend ('local' or 'elevenlabs')."""
    global _current_backend
    if backend not in ("local", "elevenlabs"):
        raise ValueError(f"Invalid backend: {backend}. Use 'local' or 'elevenlabs'")
    _current_backend = backend


def is_backend_available(backend: Optional[Backend] = None) -> bool:
    """Check if a backend is available."""
    backend = backend or _current_backend
    if backend == "local":
        return is_local_tts_available()
    elif backend == "elevenlabs":
        return _requests_available and bool(ELEVENLABS_API_KEY)
    return False


# -----------------------------------------------------------------------------
# Mute Control
# -----------------------------------------------------------------------------

def is_muted() -> bool:
    """Check if sound is muted."""
    return _muted


def set_muted(muted: bool) -> bool:
    """Set mute state."""
    global _muted
    _muted = muted
    return _muted


def mute():
    """Mute sound notifications."""
    return set_muted(True)


def unmute():
    """Unmute sound notifications."""
    return set_muted(False)


def toggle_mute() -> bool:
    """Toggle mute state."""
    return set_muted(not _muted)


# -----------------------------------------------------------------------------
# Audio Playback (Cross-Platform)
# -----------------------------------------------------------------------------

def _play_wav(filepath: str):
    """Play a WAV file synchronously (cross-platform)."""
    import subprocess
    if sys.platform == 'win32':
        ps_cmd = f'(New-Object Media.SoundPlayer "{filepath}").PlaySync()'
        subprocess.run(['powershell', '-Command', ps_cmd], check=False,
                      capture_output=True)
    elif sys.platform == 'darwin':
        subprocess.run(['afplay', filepath], check=False, capture_output=True)
    else:
        subprocess.run(['aplay', '-q', filepath], check=False, capture_output=True)


def _play_mp3(filepath: str):
    """Play an MP3 file (cross-platform)."""
    import subprocess
    import time
    if sys.platform == 'win32':
        # Try ffplay first if available (waits for completion)
        try:
            subprocess.run(['ffplay', '-nodisp', '-autoexit', '-loglevel', 'quiet', filepath],
                          check=True, capture_output=True)
            return
        except (subprocess.CalledProcessError, FileNotFoundError):
            pass
        # Fallback: use start command with default player
        subprocess.run(['cmd', '/c', 'start', '', filepath], check=False,
                      capture_output=True, shell=False)
        # Estimate wait time from file size (~24KB/sec for typical MP3)
        file_size = os.path.getsize(filepath)
        wait_time = max(1, file_size / 24000) + 0.5
        time.sleep(wait_time)
    elif sys.platform == 'darwin':
        subprocess.run(['afplay', filepath], check=False, capture_output=True)
    else:
        # Try mpv, then ffplay, then mpg123
        for cmd in [['mpv', '--no-video', filepath],
                    ['ffplay', '-nodisp', '-autoexit', filepath],
                    ['mpg123', '-q', filepath]]:
            try:
                subprocess.run(cmd, check=True, capture_output=True)
                return
            except (subprocess.CalledProcessError, FileNotFoundError):
                continue


# -----------------------------------------------------------------------------
# Tone Generation
# -----------------------------------------------------------------------------

def _generate_tone(wav, freq: float, duration_ms: int, volume: float, sample_rate: int):
    """Generate a single tone and write to wav file."""
    n_samples = int(sample_rate * duration_ms / 1000)
    for i in range(n_samples):
        t = i / sample_rate
        value = int(32767 * volume * math.sin(2 * math.pi * freq * t))
        wav.writeframes(struct.pack('<h', value))


def _generate_silence(wav, duration_ms: int, sample_rate: int):
    """Generate silence and write to wav file."""
    n_samples = int(sample_rate * duration_ms / 1000)
    wav.writeframes(b'\x00\x00' * n_samples)


def _ensure_beep_file():
    """Create the two-tone alert beep WAV file if it doesn't exist."""
    if os.path.exists(_beep_file):
        return
    tonic = 800
    major_sixth = tonic * 5 / 3
    volume = 0.5
    sample_rate = 44100
    with wave.open(_beep_file, 'w') as wav:
        wav.setnchannels(1)
        wav.setsampwidth(2)
        wav.setframerate(sample_rate)
        _generate_tone(wav, major_sixth, 70, volume, sample_rate)
        _generate_silence(wav, 15, sample_rate)
        _generate_tone(wav, tonic, 70, volume, sample_rate)


def _ensure_startup_beep_file():
    """Create the startup chime WAV file if it doesn't exist."""
    if os.path.exists(_startup_beep_file):
        return
    c5, e5, g5 = 523, 659, 784
    volume = 0.45
    sample_rate = 44100
    with wave.open(_startup_beep_file, 'w') as wav:
        wav.setnchannels(1)
        wav.setsampwidth(2)
        wav.setframerate(sample_rate)
        _generate_tone(wav, c5, 55, volume, sample_rate)
        _generate_silence(wav, 10, sample_rate)
        _generate_tone(wav, e5, 55, volume, sample_rate)
        _generate_silence(wav, 10, sample_rate)
        _generate_tone(wav, g5, 80, volume * 1.1, sample_rate)


def play_beep():
    """Play the alert beep sound."""
    if _muted:
        return
    try:
        _ensure_beep_file()
        _play_wav(_beep_file)
    except Exception:
        pass


def play_startup_chime():
    """Play the startup chime sound."""
    if _muted:
        return
    try:
        _ensure_startup_beep_file()
        _play_wav(_startup_beep_file)
    except Exception:
        pass


# -----------------------------------------------------------------------------
# Local TTS (Piper)
# -----------------------------------------------------------------------------

def is_local_tts_available() -> bool:
    """Check if local TTS (Piper) is available."""
    global _tts_available
    if _tts_available is None:
        try:
            import piper
            _tts_available = True
        except ImportError:
            _tts_available = False
    return _tts_available


def _download_voice_model(voice_path: str) -> Optional[str]:
    """Download a voice model from Hugging Face if not present."""
    try:
        from huggingface_hub import hf_hub_download
        print(f"Downloading voice model: {voice_path}...")
        model_file = hf_hub_download(
            repo_id='rhasspy/piper-voices',
            filename=voice_path,
            cache_dir=_voices_dir
        )
        hf_hub_download(
            repo_id='rhasspy/piper-voices',
            filename=voice_path + '.json',
            cache_dir=_voices_dir
        )
        print(f"Downloaded: {voice_path}")
        return model_file
    except Exception as e:
        print(f"Failed to download {voice_path}: {e}")
        return None


def _find_voice_model(voice_path: str) -> Optional[str]:
    """Find a voice model, downloading if necessary."""
    import glob
    pattern = os.path.join(_voices_dir, 'models--rhasspy--piper-voices',
                          'snapshots', '*', voice_path)
    matches = glob.glob(pattern)
    if matches:
        return matches[0]
    return _download_voice_model(voice_path)


def _load_voices() -> bool:
    """Load Piper voice models (lazy initialization, auto-downloads if needed)."""
    global _male_voice, _female_voice, _voices_loaded

    if _voices_loaded:
        return _male_voice is not None and _female_voice is not None

    if not is_local_tts_available():
        _voices_loaded = True
        return False

    try:
        from piper import PiperVoice
        male_model = _find_voice_model(_voice_models['male'])
        female_model = _find_voice_model(_voice_models['female'])

        if male_model and os.path.exists(male_model):
            _male_voice = PiperVoice.load(male_model)
        if female_model and os.path.exists(female_model):
            _female_voice = PiperVoice.load(female_model)

        _voices_loaded = True
        return _male_voice is not None and _female_voice is not None
    except Exception:
        _voices_loaded = True
        return False


def _synthesize_to_file(voice, text: str, filepath: str):
    """Synthesize text to a WAV file using Piper."""
    wav_file = wave.open(filepath, 'wb')
    wav_file.setnchannels(1)
    wav_file.setsampwidth(2)
    wav_file.setframerate(voice.config.sample_rate)
    voice.synthesize_wav(text, wav_file)
    # Add silence buffer for Windows playback
    silence_samples = int(voice.config.sample_rate * 2)
    wav_file.writeframes(b'\x00\x00' * silence_samples)
    wav_file.close()


def _speak_local(text: str, voice: str = "male") -> bool:
    """Speak text using local Piper TTS."""
    with _voice_lock:
        if not _load_voices():
            return False

        target_voice = _male_voice if voice == "male" else _female_voice
        if target_voice is None:
            return False

        try:
            with tempfile.NamedTemporaryFile(suffix='.wav', delete=False) as f:
                temp_path = f.name
            _synthesize_to_file(target_voice, text, temp_path)
            _play_wav(temp_path)
            os.unlink(temp_path)
            return True
        except Exception as e:
            print(f"Local TTS error: {e}")
            return False


# -----------------------------------------------------------------------------
# ElevenLabs TTS
# -----------------------------------------------------------------------------

def _speak_elevenlabs(text: str, voice_id: Optional[str] = None,
                      voice_settings: Optional[dict] = None) -> bool:
    """Speak text using ElevenLabs API."""
    if not _requests_available:
        print("ElevenLabs requires 'requests' package")
        return False

    if not ELEVENLABS_API_KEY:
        print("ElevenLabs API key not configured")
        return False

    voice_id = voice_id or ELEVENLABS_VOICE_ID
    settings = voice_settings or ELEVENLABS_VOICE_SETTINGS

    url = f"https://api.elevenlabs.io/v1/text-to-speech/{voice_id}"

    headers = {
        "xi-api-key": ELEVENLABS_API_KEY,
        "Content-Type": "application/json"
    }

    data = {
        "text": text,
        "model_id": ELEVENLABS_MODEL_ID,
        "voice_settings": settings
    }

    try:
        response = requests.post(url, headers=headers, json=data, timeout=30)

        if response.status_code == 200:
            with tempfile.NamedTemporaryFile(suffix='.mp3', delete=False) as f:
                f.write(response.content)
                temp_path = f.name
            _play_mp3(temp_path)
            try:
                os.unlink(temp_path)
            except OSError:
                pass  # File may still be in use by async player
            return True
        else:
            print(f"ElevenLabs error: {response.status_code}")
            print(response.text)
            return False
    except Exception as e:
        print(f"ElevenLabs error: {e}")
        return False


def elevenlabs_generate_to_file(text: str, output_path: str,
                                 voice_id: Optional[str] = None,
                                 voice_settings: Optional[dict] = None) -> bool:
    """Generate audio to file using ElevenLabs (no playback)."""
    if not _requests_available:
        print("ElevenLabs requires 'requests' package")
        return False

    if not ELEVENLABS_API_KEY:
        print("ElevenLabs API key not configured")
        return False

    voice_id = voice_id or ELEVENLABS_VOICE_ID
    settings = voice_settings or ELEVENLABS_VOICE_SETTINGS

    url = f"https://api.elevenlabs.io/v1/text-to-speech/{voice_id}"

    headers = {
        "xi-api-key": ELEVENLABS_API_KEY,
        "Content-Type": "application/json"
    }

    data = {
        "text": text,
        "model_id": ELEVENLABS_MODEL_ID,
        "voice_settings": settings
    }

    try:
        response = requests.post(url, headers=headers, json=data, timeout=30)

        if response.status_code == 200:
            with open(output_path, 'wb') as f:
                f.write(response.content)
            return True
        else:
            print(f"ElevenLabs error: {response.status_code}")
            print(response.text)
            return False
    except Exception as e:
        print(f"ElevenLabs error: {e}")
        return False


# -----------------------------------------------------------------------------
# Main API
# -----------------------------------------------------------------------------

def speak(text: str, backend: Optional[Backend] = None, force: bool = False,
          voice: str = "male", **kwargs) -> bool:
    """
    Speak text using TTS.

    Args:
        text: Text to speak
        backend: 'local' or 'elevenlabs' (uses current backend if not specified)
        force: Speak even if muted
        voice: For local TTS: 'male' or 'female'
        **kwargs: Additional args passed to backend (e.g., voice_id for elevenlabs)

    Returns:
        True if successful, False otherwise
    """
    if _muted and not force:
        return False

    backend = backend or _current_backend

    if backend == "local":
        return _speak_local(text, voice=voice)
    elif backend == "elevenlabs":
        return _speak_elevenlabs(text, **kwargs)
    else:
        print(f"Unknown backend: {backend}")
        return False


def speak_async(text: str, backend: Optional[Backend] = None, force: bool = False,
                voice: str = "male", **kwargs):
    """Speak text in a background thread (non-blocking)."""
    if _muted and not force:
        return
    thread = threading.Thread(
        target=speak,
        args=(text,),
        kwargs={"backend": backend, "force": force, "voice": voice, **kwargs},
        daemon=True
    )
    thread.start()
    return thread


def announce(prefix: str, message: str, backend: Optional[Backend] = None,
             blocking: bool = False):
    """
    Make a dual-part announcement (prefix + message).

    For local TTS: Uses male voice for prefix, female for message.
    For ElevenLabs: Uses single voice for both.
    """
    if _muted:
        return

    backend = backend or _current_backend

    def _do_announce():
        play_beep()
        if backend == "local":
            _speak_local(prefix, voice="male")
            _speak_local(message, voice="female")
        else:
            _speak_elevenlabs(f"{prefix} {message}")

    if blocking:
        _do_announce()
    else:
        thread = threading.Thread(target=_do_announce, daemon=True)
        thread.start()


# -----------------------------------------------------------------------------
# Convenience Functions
# -----------------------------------------------------------------------------

def say(text: str, **kwargs) -> bool:
    """Alias for speak()."""
    return speak(text, **kwargs)


def say_async(text: str, **kwargs):
    """Alias for speak_async()."""
    return speak_async(text, **kwargs)


# -----------------------------------------------------------------------------
# Configuration Helpers
# -----------------------------------------------------------------------------

def configure_elevenlabs(api_key: Optional[str] = None,
                         voice_id: Optional[str] = None,
                         model_id: Optional[str] = None,
                         voice_settings: Optional[dict] = None):
    """Configure ElevenLabs settings."""
    global ELEVENLABS_API_KEY, ELEVENLABS_VOICE_ID, ELEVENLABS_MODEL_ID
    global ELEVENLABS_VOICE_SETTINGS

    if api_key:
        ELEVENLABS_API_KEY = api_key
    if voice_id:
        ELEVENLABS_VOICE_ID = voice_id
    if model_id:
        ELEVENLABS_MODEL_ID = model_id
    if voice_settings:
        ELEVENLABS_VOICE_SETTINGS.update(voice_settings)


def get_status() -> dict:
    """Get status of voice module."""
    return {
        "backend": _current_backend,
        "muted": _muted,
        "local_available": is_local_tts_available(),
        "elevenlabs_available": _requests_available and bool(ELEVENLABS_API_KEY),
        "elevenlabs_voice_id": ELEVENLABS_VOICE_ID,
        "elevenlabs_model_id": ELEVENLABS_MODEL_ID,
    }