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",
    "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",
}

MACOS_VOICES = {
    "Moira": "Irish Female",
    "Daniel": "UK Male",
    "Samantha": "US Female",
    "Karen": "Australian Female",
    "Alex": "US 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",
}


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 apply_voice(self, voice_name: str) -> tuple[bool, str]:
        voice_name = voice_name.strip()
        matched_voice = None

        # Check voice aliases first (e.g. Daniel -> bm_george)
        clean_name = voice_name.lower()
        if clean_name in VOICE_ALIASES:
            matched_voice = VOICE_ALIASES[clean_name]

        # Exact match or fuzzy match
        if not matched_voice:
            for v in KOKORO_VOICES:
                if clean_name == v.lower():
                    matched_voice = v
                    break

        if not matched_voice:
            # Partial match search
            for v in KOKORO_VOICES:
                if clean_name in v.lower():
                    matched_voice = v
                    break

        if not matched_voice:
            available = ", ".join(list(KOKORO_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}."