handler.py
Browse files- handler.py +0 -66
handler.py
DELETED
@@ -1,66 +0,0 @@
|
|
1 |
-
from pydantic import BaseModel
|
2 |
-
from environs import Env
|
3 |
-
from typing import List, Dict, Any
|
4 |
-
import os
|
5 |
-
import base64
|
6 |
-
import numpy as np
|
7 |
-
import librosa
|
8 |
-
from scipy.io import wavfile
|
9 |
-
import asyncio
|
10 |
-
|
11 |
-
class EndpointHandler:
|
12 |
-
def __init__(self, model_dir=None):
|
13 |
-
self.model_dir = model_dir
|
14 |
-
|
15 |
-
def __call__(self, data: Dict[str, Any]) -> Dict[str, Any]:
|
16 |
-
try:
|
17 |
-
# Clone the repository
|
18 |
-
repo_url = "https://huggingface.co/mazalaai/TTS_Mongolian.git"
|
19 |
-
os.system(f"git clone {repo_url}")
|
20 |
-
|
21 |
-
# Change directory to the cloned repository
|
22 |
-
repo_dir = "TTS_Mongolian"
|
23 |
-
os.chdir(repo_dir)
|
24 |
-
|
25 |
-
# Import the voice_processing module and functions
|
26 |
-
from voice_processing import tts, get_model_names, voice_mapping, get_unique_filename
|
27 |
-
|
28 |
-
if "inputs" in data: # Check if data is in Hugging Face JSON format
|
29 |
-
return self.process_hf_input(data)
|
30 |
-
else:
|
31 |
-
return self.process_json_input(data)
|
32 |
-
except ValueError as e:
|
33 |
-
return {"error": str(e)}
|
34 |
-
except Exception as e:
|
35 |
-
return {"error": str(e)}
|
36 |
-
|
37 |
-
def process_json_input(self, json_data):
|
38 |
-
if all(key in json_data for key in ["model_name", "tts_text", "selected_voice", "slang_rate", "use_uploaded_voice"]):
|
39 |
-
model_name = json_data["model_name"]
|
40 |
-
tts_text = json_data["tts_text"]
|
41 |
-
selected_voice = json_data["selected_voice"]
|
42 |
-
slang_rate = json_data["slang_rate"]
|
43 |
-
use_uploaded_voice = json_data["use_uploaded_voice"]
|
44 |
-
voice_upload_file = json_data.get("voice_upload_file", None)
|
45 |
-
|
46 |
-
edge_tts_voice = voice_mapping.get(selected_voice)
|
47 |
-
if not edge_tts_voice:
|
48 |
-
raise ValueError(f"Invalid voice '{selected_voice}'.")
|
49 |
-
|
50 |
-
info, edge_tts_output_path, tts_output_data, edge_output_file = asyncio.run(tts(
|
51 |
-
model_name, tts_text, edge_tts_voice, slang_rate, use_uploaded_voice, voice_upload_file
|
52 |
-
))
|
53 |
-
|
54 |
-
if edge_output_file and os.path.exists(edge_output_file):
|
55 |
-
|
56 |
-
def process_hf_input(self, hf_data):
|
57 |
-
if "inputs" in hf_data:
|
58 |
-
actual_data = hf_data["inputs"]
|
59 |
-
return self.process_json_input(actual_data)
|
60 |
-
else:
|
61 |
-
return {"error": "Invalid Hugging Face JSON structure."}
|
62 |
-
|
63 |
-
def save_audio_data_to_file(self, audio_data, sample_rate=40000):
|
64 |
-
file_path = get_unique_filename('wav')
|
65 |
-
wavfile.write(file_path, sample_rate, audio_data)
|
66 |
-
return file_path
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|