| from typing import Dict, List, Any |
| import base64 |
| import tempfile |
| import os |
| import subprocess |
| import sys |
|
|
| class EndpointHandler: |
| def __init__(self, path=""): |
| print("Starting GPU Cloud Environment... (The Monkey Patch Fix)") |
| |
| try: |
| |
| print("Installing OS audio drivers...") |
| subprocess.check_call("apt-get update && apt-get install -y libsndfile1 ffmpeg", shell=True) |
| |
| |
| print("Installing TTS while locking PyTorch versions...") |
| subprocess.check_call([ |
| sys.executable, "-m", "pip", "install", |
| "TTS==0.22.0", "torch==2.5.1", "torchaudio==2.5.1", |
| "--ignore-installed", "blinker", "--no-cache-dir" |
| ]) |
| |
| |
| print("Forcing clean Scipy binary...") |
| subprocess.check_call([ |
| sys.executable, "-m", "pip", "install", "--force-reinstall", |
| "--only-binary", "scipy", "scipy==1.13.0", "--no-cache-dir" |
| ]) |
| print("Dependencies safely locked and installed!") |
| except Exception as e: |
| print(f"Installation crashed: {e}") |
| |
| |
| import torch |
| from TTS.api import TTS |
| |
| |
| |
| TTS.is_multi_lingual = property(lambda self: False) |
| TTS.is_multi_speaker = property(lambda self: False) |
| |
| |
| print("Loading Ormuri Voice into the Cloud GPU...") |
| |
| model_file = os.path.join(path, "best_model.pth") |
| config_file = os.path.join(path, "config.json") |
| |
| |
| self.tts = TTS(model_path=model_file, config_path=config_file, progress_bar=False, gpu=True) |
| |
| print("GPU TTS Engine Ready!") |
|
|
| def __call__(self, data: Dict[str, Any]) -> List[Dict[str, Any]]: |
| text = data.pop("inputs", "") |
| |
| with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as temp_audio: |
| temp_path = temp_audio.name |
|
|
| self.tts.tts_to_file(text=text, file_path=temp_path) |
|
|
| with open(temp_path, "rb") as f: |
| audio_bytes = f.read() |
| |
| base64_audio = base64.b64encode(audio_bytes).decode("utf-8") |
| os.remove(temp_path) |
|
|
| return [{"audio": base64_audio, "sampling_rate": 22050}] |