VoiceAgent/voice_manager.py

"""Voice Manager for listing and dynamically changing TTS voices at runtime.

Supports both Kokoro TTS voices and macOS System voices, persisting user voice preferences to disk.
"""

import json
import logging
from pathlib import Path

try:
    from loguru import logger
except ImportError:
    logger = logging.getLogger("voice_manager")

# Supported Kokoro TTS voices categorized by style
KOKORO_VOICES = {
    "af_heart": "American Female - Warm & Natural (Default)",
    "af_bella": "American Female - Clear & Expressive",
    "af_sarah": "American Female - Soft & Smooth",
    "af_nicole": "American Female - Relaxed",
    "af_sky": "American Female - Bright",
    "af_alba": "American Female - Pocket Alba",
    "am_michael": "American Male - Friendly & Crisp",
    "am_adam": "American Male - Natural",
    "am_fenrir": "American Male - Deep",
    "am_puck": "American Male - Energetic",
    "bf_emma": "British Female - Professional",
    "bf_isabella": "British Female - Smooth",
    "bm_george": "British Male - Warm",
    "bm_fable": "British Male - Expressive",
    "bm_stuart": "British Male - Stuart Bell",
    "custom_pocket": "Pocket Voice Clone (Default)",
    "jv_pocket": "JV Voice Profile (Kyutai Pocket TTS)",
    "qwen_jv": "JV Voice Profile (MLX Qwen 1.7B TTS)",
}

MACOS_VOICES = {
    "Moira": "Irish Female",
    "Daniel": "UK Male",
    "Samantha": "US Female",
    "Karen": "Australian Female",
    "Alex": "US Male",
    "Stuart": "UK Male",
}

VOICE_ALIASES = {
    "daniel": "bm_george",      # British Male
    "moira": "bf_emma",         # British Female
    "samantha": "af_bella",     # US Female
    "alex": "am_michael",       # US Male
    "karen": "bf_isabella",     # AU/UK Female
    "michael": "am_michael",
    "bella": "af_bella",
    "heart": "af_heart",
    "fenrir": "am_fenrir",
    "adam": "am_adam",
    "alba": "af_alba",
    "pocket alba": "af_alba",
    "pocketalba": "af_alba",
    "pocket_alba": "af_alba",
    "stuart": "bm_george",
    "stuart bell": "bm_george",
    "stuartbell": "bm_george",
    "stuart_bell": "bm_george",
    "bell": "bm_george",
    "custom_pocket": "custom_pocket",
    "pocket_custom": "custom_pocket",
    "pocket custom": "custom_pocket",
    "custom voice": "custom_pocket",
    "jv": "jv_pocket",
    "jv_pocket": "jv_pocket",
    "jv pocket": "jv_pocket",
    "jv profile": "jv_pocket",
    "jv voice": "jv_pocket",
    "qwen": "qwen_jv",
    "qwen_jv": "qwen_jv",
    "qwen 1.7b": "qwen_jv",
    "qwen3": "qwen_jv",
}


class VoiceManager:
    """Manages active voice settings and dynamic voice switching."""

    def __init__(self, workspace_dir: Path | None = None, tts_processor=None):
        self._workspace_dir = Path(workspace_dir) if workspace_dir else Path(__file__).parent
        self._tts_processor = tts_processor
        self._config_file = self._workspace_dir / "voice_settings.json"
        self._active_voice = "af_heart"
        self.load_saved_voice()

    def set_tts_processor(self, tts_processor):
        self._tts_processor = tts_processor
        if self._active_voice:
            self.apply_voice(self._active_voice)

    def load_saved_voice(self) -> str:
        if self._config_file.exists():
            try:
                data = json.loads(self._config_file.read_text())
                if "voice" in data:
                    self._active_voice = data["voice"]
                    logger.info(f"Loaded saved voice preference: {self._active_voice}")
            except Exception as e:
                logger.warning(f"Could not load saved voice settings: {e}")
        return self._active_voice

    def sync_voice(self) -> str:
        """Check if voice_settings.json was updated on disk and update TTS live."""
        current_disk_voice = self.load_saved_voice()
        if self._tts_processor and current_disk_voice:
            if hasattr(self._tts_processor, "_settings"):
                current_tts_voice = getattr(self._tts_processor._settings, "voice", None)
                if current_tts_voice != current_disk_voice:
                    self._tts_processor._settings.voice = current_disk_voice
                    logger.info(f"Live TTS voice synced to: {current_disk_voice}")
            elif hasattr(self._tts_processor, "set_voice"):
                self._tts_processor.set_voice(current_disk_voice)
        return current_disk_voice

    def save_voice(self, voice_name: str):
        try:
            self._config_file.write_text(json.dumps({"voice": voice_name}, indent=2))
        except Exception as e:
            logger.warning(f"Could not save voice setting: {e}")

    def list_voices(self) -> str:
        kokoro_lines = [f"- {v}: {desc}" for v, desc in KOKORO_VOICES.items()]
        macos_lines = [f"- {v}: {desc}" for v, desc in MACOS_VOICES.items()]
        return (
            "Available Voices:\n\n"
            "Kokoro Voices:\n" + "\n".join(kokoro_lines) + "\n\n"
            "macOS System Voices:\n" + "\n".join(macos_lines)
        )

    def set_voice(self, voice_name: str) -> tuple[bool, str]:
        return self.apply_voice(voice_name)

    def apply_voice(self, voice_name: str) -> tuple[bool, str]:
        import re
        voice_name = voice_name.strip()
        matched_voice = None

        clean_name = voice_name.lower()
        norm_name = re.sub(r"[^a-z0-9]", "", clean_name)

        # Check voice aliases first (e.g. Daniel -> bm_george, bella -> af_bella)
        if clean_name in VOICE_ALIASES:
            matched_voice = VOICE_ALIASES[clean_name]
        elif norm_name in VOICE_ALIASES:
            matched_voice = VOICE_ALIASES[norm_name]

        # Exact or normalized match in Kokoro voices (e.g. afbella -> af_bella)
        if not matched_voice:
            for v in KOKORO_VOICES:
                v_norm = re.sub(r"[^a-z0-9]", "", v.lower())
                if norm_name == v_norm or clean_name == v.lower():
                    matched_voice = v
                    break

        if not matched_voice:
            # Partial/substring match search in Kokoro voices
            for v in KOKORO_VOICES:
                v_norm = re.sub(r"[^a-z0-9]", "", v.lower())
                if norm_name in v_norm or v_norm in norm_name:
                    matched_voice = v
                    break

        if not matched_voice:
            # Match in macOS voices
            for v in MACOS_VOICES:
                v_norm = re.sub(r"[^a-z0-9]", "", v.lower())
                if norm_name == v_norm or clean_name == v.lower() or norm_name in v_norm or v_norm in norm_name:
                    matched_voice = v
                    break

        if not matched_voice:
            available = ", ".join(list(KOKORO_VOICES.keys()) + list(MACOS_VOICES.keys()))
            return False, f"Voice '{voice_name}' not found. Available voices: {available}"

        self._active_voice = matched_voice
        self.save_voice(matched_voice)

        if self._tts_processor:
            try:
                # Update Kokoro TTS setting if active
                if hasattr(self._tts_processor, "_settings"):
                    self._tts_processor._settings.voice = matched_voice
                    logger.info(f"Dynamic voice updated to: {matched_voice}")
                    return True, f"Voice changed to {matched_voice}."
                elif hasattr(self._tts_processor, "set_voice"):
                    self._tts_processor.set_voice(matched_voice)
                    logger.info(f"Dynamic voice updated to: {matched_voice}")
                    return True, f"Voice changed to {matched_voice}."
            except Exception as e:
                logger.error(f"Failed to apply voice to TTS processor: {e}")
                return False, f"Could not change voice: {e}"

        return True, f"Voice set to {matched_voice}."