handler.py

import os
import base64
import numpy as np
import librosa
from scipy.io import wavfile
from typing import Optional
from voice_processing import tts, get_model_names, voice_mapping, get_unique_filename

class EndpointHandler:
    def __init__(self):
        pass

    def get_models(self):
        return get_model_names()

    def get_voices(self):
        return list(voice_mapping.keys())

    def save_audio_data_to_file(self, audio_data, sample_rate=40000):
        file_path = get_unique_filename('wav')  # Generate a unique file name
        wavfile.write(file_path, sample_rate, audio_data)
        return file_path

    async def convert_tts(self, model_name: str, tts_text: str = "Текстыг оруулна уу.", selected_voice: str = "", slang_rate: float = 0.0, use_uploaded_voice: bool = False, voice_upload: Optional[bytes] = None):
        edge_tts_voice = voice_mapping.get(selected_voice)
        if not edge_tts_voice:
            raise ValueError(f"Invalid voice '{selected_voice}'.")

        if use_uploaded_voice and voice_upload is None:
            raise ValueError("No voice file uploaded")

        # Process the text input or uploaded voice
        info, edge_tts_output_path, tts_output_data, edge_output_file = await tts(
            model_name, tts_text, edge_tts_voice, slang_rate, use_uploaded_voice, voice_upload
        )

        if edge_output_file and os.path.exists(edge_output_file):
            os.remove(edge_output_file)

        _, audio_output = tts_output_data

        # Generate a unique filename and save the audio data
        audio_file_path = self.save_audio_data_to_file(audio_output) if isinstance(audio_output, np.ndarray) else audio_output

        # Encode the audio file to base64
        try:
            with open(audio_file_path, 'rb') as file:
                audio_bytes = file.read()
            audio_data_uri = f"data:audio/wav;base64,{base64.b64encode(audio_bytes).decode('utf-8')}"
        except Exception as e:
            raise ValueError(f"Failed to read audio file: {e}")
        finally:
            # Cleanup the temporary audio file
            if os.path.exists(audio_file_path):
                os.remove(audio_file_path)

        return {"info": info, "audio_data_uri": audio_data_uri}

    def convert_to_audio_bytes(self, audio_file_path):
        try:
            with open(audio_file_path, 'rb') as audio_file:
                return audio_file.read()
        except Exception as e:
            print(f"Error reading audio file: {e}")
            return None