backend/voice/tts.py

# backend/voice/tts.py
# §0.4 — Single entrypoint. ALL existing callers remain byte-for-byte identical.
# JARVIS routes through 3-engine system (Kokoro/XTTS/Chatterbox) based on context.
# FRIDAY path completely untouched.
import os
os.environ["COQUI_TOS_AGREED"] = "1"

def wav_to_bytes(wav_data) -> bytes:
    import io
    import scipy.io.wavfile
    import numpy as np
    bytes_io = io.BytesIO()
    if wav_data.dtype != np.int16:
        wav_data = np.int16(wav_data * 32767)
    scipy.io.wavfile.write(bytes_io, 24000, wav_data)
    return bytes_io.getvalue()

class TTSPipeline:
    # Legacy voice params kept for FRIDAY path — unchanged
    FRIDAY_VOICE_PARAMS = {"speed": 1.05, "pitch_shift": 0}

    _instance = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super(TTSPipeline, cls).__new__(cls)
            cls._instance._initialized = False
        return cls._instance

    def __init__(self):
        if self._initialized:
            return
        self.tts = None
        self._initialized = True

    async def synthesize(self, text: str, personality: str, language: str = "en", context: str = "conversation") -> bytes:
        """
        Single entrypoint — ZERO call-site changes required.
        FRIDAY path completely unchanged.
        JARVIS path perfectly mapped to the Ultimate Voice Clone (Kokoro + RVC).
        """
        import logging

        # ── FRIDAY path — completely untouched, existing logic, zero changes ──
        if personality != "jarvis":
            return await self._synthesize_friday(text, personality, language)

        # ── JARVIS path — 100% Ultimate Voice Clone ──────────────────────────
        logging.info(f"[TTS] JARVIS → Ultimate Voice Clone (Kokoro+RVC), context={context}")

        # Supported XTTS/Kokoro languages (Edge-TTS fallback for others)
        supported_langs = {"en", "es", "fr", "de", "it", "pt", "pl", "tr", "ru", "nl", "cs", "ar", "zh-cn", "hu", "ko", "ja", "hi"}
        if language == "zh": language = "zh-cn"

        # Non-supported language → always Edge-TTS
        if language not in supported_langs:
            return await self._synthesize_edge_tts(text, personality, language)

        # REGRESSION GUARD:
        # Any new code calling TTS for persona="jarvis" must go through synthesize_tts().
        # Never call kokoro_engine, rvc_infer, or any DSP function directly from outside tts.py.
        # CI Check equivalent:
        # grep -rn "kokoro_engine\.\|rvc_infer\|dsp_vp4\." backend/ --include="*.py" | grep -v "backend/voice/tts.py\|backend/voice/kokoro_engine.py\|backend/voice/dsp_vp4.py"
        from backend.voice.engines.kokoro_engine import synthesize_kokoro
        return await synthesize_kokoro(text)

    async def _synthesize_edge_tts(self, text: str, personality: str, language: str) -> bytes:
        """Edge-TTS fallback for languages outside XTTS-v2's 17-language support."""
        import logging
        logging.info(f"[TTS] Routing to Edge-TTS for unsupported language: '{language}'")
        try:
            import edge_tts
            target_gender = "Male" if personality == "jarvis" else "Female"
            selected_voice = None
            # For English, honour the identity file's chosen persona voice
            # (JARVIS → British male, FRIDAY → Irish female) instead of grabbing
            # the first gender match, so the cloud voice matches the desktop one.
            if language.lower().startswith("en"):
                try:
                    from modules.assistant_identity import get_voice_id_for_mode
                    selected_voice = get_voice_id_for_mode("jarvis" if personality == "jarvis" else "friday")
                except Exception:
                    selected_voice = None
            if not selected_voice:
                voices = await edge_tts.list_voices()
                for v in voices:
                    if v["Locale"].lower().startswith(language.lower()) and v["Gender"] == target_gender:
                        selected_voice = v["ShortName"]
                        break
            if not selected_voice:
                selected_voice = "gu-IN-NiranjanNeural" if target_gender == "Male" else "gu-IN-DhwaniNeural"
            communicate = edge_tts.Communicate(text, selected_voice)
            audio_bytes = b""
            async for chunk in communicate.stream():
                if chunk["type"] == "audio":
                    audio_bytes += chunk["data"]
            import io, librosa, scipy.io.wavfile, numpy as np
            y, sr = librosa.load(io.BytesIO(audio_bytes), sr=24000)
            pcm = np.int16(y * 32767)
            out = io.BytesIO()
            scipy.io.wavfile.write(out, 24000, pcm)
            return out.getvalue()
        except Exception as e:
            logging.error(f"[TTS] Edge-TTS failed: {e}. Falling back to default English engine.")
            from backend.voice.engines.kokoro_engine import synthesize_kokoro
            return await synthesize_kokoro(text)

    async def _synthesize_friday(self, text: str, personality: str, language: str) -> bytes:
        """FRIDAY TTS path — completely unchanged from original implementation."""
        supported_langs = {"en", "es", "fr", "de", "it", "pt", "pl", "tr", "ru", "nl", "cs", "ar", "zh-cn", "hu", "ko", "ja", "hi"}
        if language not in supported_langs:
            return await self._synthesize_edge_tts(text, personality, language)

        import os, asyncio
        app_data = os.environ.get("JARVIS_APP_DATA_DIR", ".")
        fallback_wav = os.path.join(app_data, "voices", f"{personality}.wav")
        speaker_wav = os.environ.get("FRIDAY_VOICE_SAMPLE_PATH", fallback_wav)
        params = self.FRIDAY_VOICE_PARAMS

        loop = asyncio.get_running_loop()
        def _sync_tts():
            if self.tts is None:
                from TTS.api import TTS as CoquiTTS
                self.tts = CoquiTTS("tts_models/multilingual/multi-dataset/xtts_v2")
            return self.tts.tts(text=text, language=language, speaker_wav=speaker_wav, **params)
        try:
            wav = await loop.run_in_executor(None, _sync_tts)
        except Exception as e:
            # Coqui XTTS is intentionally NOT installed on the cloud Space (too
            # heavy), and the speaker clone wav may be absent there too. Rather
            # than 500 (which left the DEFAULT persona with no cloud voice at
            # all), fall back to Edge-TTS in FRIDAY's identity voice.
            import logging
            logging.warning(f"[TTS] FRIDAY XTTS unavailable ({e}); falling back to Edge-TTS.")
            return await self._synthesize_edge_tts(text, personality, language)
        res = wav_to_bytes(wav)
        del wav
        import gc
        gc.collect()
        return res