Spaces:
Running on Zero
Running on Zero
| """Voice engine — STT (Whisper) and TTS (Kokoro/XTTS) with ZeroGPU.""" | |
| import logging | |
| import tempfile | |
| from pathlib import Path | |
| from typing import Optional | |
| from config import config | |
| from models.zerogpu import requires_gpu | |
| logger = logging.getLogger("synapse.voice") | |
| class VoiceEngine: | |
| """Speech-to-text and text-to-speech engine.""" | |
| def __init__(self): | |
| self._whisper_model = None | |
| self._tts_model = None | |
| def _load_whisper(self): | |
| if self._whisper_model is None: | |
| try: | |
| import torch | |
| from transformers import pipeline | |
| device = 0 if torch.cuda.is_available() else -1 | |
| self._whisper_model = pipeline( | |
| "automatic-speech-recognition", | |
| model=config.voice.whisper_model, | |
| device=device, | |
| ) | |
| logger.info("Whisper model loaded") | |
| except Exception as e: | |
| logger.warning(f"Failed to load Whisper: {e}") | |
| def _load_tts(self): | |
| if self._tts_model is None: | |
| try: | |
| self._tts_model = "kokoro" | |
| logger.info("TTS engine ready (Kokoro)") | |
| except Exception as e: | |
| logger.warning(f"Failed to load TTS: {e}") | |
| def speech_to_text(self, audio_path: str, language: str = "en") -> str: | |
| """Transcribe audio to text using Whisper.""" | |
| self._load_whisper() | |
| if self._whisper_model is None: | |
| return "[Speech recognition unavailable — model not loaded]" | |
| try: | |
| result = self._whisper_model(audio_path) | |
| return result.get("text", "") | |
| except Exception as e: | |
| logger.error(f"STT error: {e}") | |
| return f"[STT error: {e}]" | |
| def text_to_speech(self, text: str, voice: str = "default", | |
| speed: float = None) -> Optional[str]: | |
| """Convert text to speech audio file.""" | |
| self._load_tts() | |
| speed = speed or config.voice.voice_speed | |
| try: | |
| if self._tts_model == "kokoro": | |
| return self._kokoro_tts(text, voice, speed) | |
| return self._fallback_tts(text) | |
| except Exception as e: | |
| logger.error(f"TTS error: {e}") | |
| return None | |
| def _kokoro_tts(self, text: str, voice: str, speed: float) -> Optional[str]: | |
| try: | |
| from kokoro import KPipeline | |
| pipeline = KPipeline(lang_code="a") | |
| generator = pipeline(text, voice=voice, speed=speed) | |
| import soundfile as sf | |
| import numpy as np | |
| audio_chunks = [] | |
| for _, _, audio in generator: | |
| audio_chunks.append(audio) | |
| if audio_chunks: | |
| full_audio = np.concatenate(audio_chunks) | |
| output_path = Path(tempfile.mktemp(suffix=".wav", dir=str(config.DATA_DIR if hasattr(config, 'DATA_DIR') else "/tmp"))) | |
| sf.write(str(output_path), full_audio, 24000) | |
| return str(output_path) | |
| return None | |
| except Exception as e: | |
| logger.warning(f"Kokoro TTS failed: {e}") | |
| return self._fallback_tts(text) | |
| def _fallback_tts(self, text: str) -> Optional[str]: | |
| try: | |
| import subprocess | |
| output_path = Path(tempfile.mktemp(suffix=".wav", dir="/tmp")) | |
| subprocess.run( | |
| ["espeak", "-w", str(output_path), text[:500]], | |
| capture_output=True, timeout=10, | |
| ) | |
| return str(output_path) if output_path.exists() else None | |
| except Exception: | |
| return None | |
| def get_voices(self) -> list[dict]: | |
| return [ | |
| {"id": "default", "name": "Default", "language": "en"}, | |
| {"id": "af_heart", "name": "Heart", "language": "en"}, | |
| {"id": "af_bella", "name": "Bella", "language": "en"}, | |
| {"id": "am_adam", "name": "Adam", "language": "en"}, | |
| {"id": "am_michael", "name": "Michael", "language": "en"}, | |
| ] | |
| voice_engine = VoiceEngine() | |