File size: 3,770 Bytes
3ba590d 79594c2 3ba590d 79594c2 830d400 79594c2 830d400 79594c2 3ba590d 79594c2 3ba590d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 | import io
import base64
import logging
import subprocess
import sys
import numpy as np
import soundfile as sf
logger = logging.getLogger(__name__)
def _install_deps():
"""Install OmniVoice and voicetut-tts with --no-deps to avoid conflicts
with the HF Inference Endpoint base image.
Then upgrade transformers (WITH deps) because OmniVoice needs
HiggsAudioV2TokenizerModel which only exists in transformers>=4.50.
The HF Inference Toolkit already loaded successfully with the old
transformers at this point, so upgrading is safe."""
# 1. Install OmniVoice and voicetut-tts without their dep trees
pkgs = [
"git+https://github.com/k2-fsa/OmniVoice.git",
"voicetut-tts>=0.1.0",
]
for pkg in pkgs:
try:
subprocess.check_call(
[sys.executable, "-m", "pip", "install", "--no-deps", pkg],
stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE,
)
logger.info(f"Installed {pkg}")
except subprocess.CalledProcessError as e:
stderr = e.stderr.decode() if e.stderr else ""
if "already satisfied" not in stderr.lower():
logger.warning(f"Failed to install {pkg}: {stderr[:200]}")
# 2. Upgrade transformers to get HiggsAudioV2TokenizerModel (needed by OmniVoice)
try:
subprocess.check_call(
[sys.executable, "-m", "pip", "install", "--upgrade",
"transformers>=4.50.0", "tokenizers>=0.21.0"],
stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE,
)
logger.info("Upgraded transformers for OmniVoice compatibility")
except subprocess.CalledProcessError as e:
stderr = e.stderr.decode() if e.stderr else ""
logger.warning(f"Failed to upgrade transformers: {stderr[:200]}")
# Install deps before importing voicetut_tts
_install_deps()
from voicetut_tts import VoiceTutTTS, GenerationParams
class EndpointHandler:
def __init__(self, model_dir=""):
logger.info(f"Loading VoiceTut-TTS from {model_dir or 'mohammedaly22/VoiceTut-TTS'}...")
self.tts = VoiceTutTTS.from_pretrained(
model_dir or "mohammedaly22/VoiceTut-TTS",
dtype="float16",
)
self.sampling_rate = self.tts.sampling_rate
logger.info(f"VoiceTut-TTS loaded. Sampling rate: {self.sampling_rate}")
def __call__(self, data: dict) -> dict:
inputs = data.get("inputs", data)
text = inputs.get("text", "").strip()
if not text:
return {"error": "text is required"}
voice = inputs.get("voice", "Mohamed")
language = inputs.get("language", "arz")
normalize = inputs.get("normalize", True)
num_step = int(inputs.get("num_step", 32))
guidance_scale = float(inputs.get("guidance_scale", 2.0))
speed = float(inputs.get("speed", 1.0))
params = GenerationParams(
num_step=num_step,
guidance_scale=guidance_scale,
speed=speed,
)
try:
wav = self.tts.synthesize(
text,
speaker=voice,
language=language,
normalize=normalize,
params=params,
)
except Exception as e:
logger.exception("Synthesis failed")
return {"error": str(e)}
buf = io.BytesIO()
sf.write(buf, wav, self.sampling_rate, format="WAV")
wav_bytes = buf.getvalue()
wav_b64 = base64.b64encode(wav_bytes).decode("utf-8")
return {
"audio_base64": wav_b64,
"sampling_rate": self.sampling_rate,
"duration_sec": float(len(wav) / self.sampling_rate),
} |