app.py

import json
import mimetypes
import os
import shutil
import tempfile
import time

import assemblyai as aai
import gradio as gr
import numpy as np
import requests
import sounddevice as sd
from elevenlabs import clone, generate, play, set_api_key, stream
from scipy.io.wavfile import write

set_api_key("cedcbf1991539f9c825a9346e1b7b708")
import mimetypes

from gradio.components import Audio, Radio, Textbox
from gradio.components import Audio as AudioInput
from gradio.components import Audio as AudioOutput
from gradio.components import Textbox as TextboxOutput

APP_KEY = "6lWL15cmmm5y5hLYU8-MvQ=="
APP_SECRET = "xoXvx_qwuD5HczjnEYOC9OJj6HGCZDFZBHKHEegigHA="

aai.settings.api_key = "6c7f4d60028e4df9b889b93acb8ed698"


def transcribe_audio(file_path):
    transcriber = aai.Transcriber()
    transcript = transcriber.transcribe(file_path)
    return transcript.text


def clone_and_stream_voice(name, description, labels, text, model):
    voice = clone(
        name=name, description=description, files=["output.wav"], labels=labels
    )

    audio = generate(
        text=text,
        voice=voice,
        model=model,
        stream=True,
        stream_chunk_size=2048,
        latency=1,
    )

    stream(audio)


def get_access_token():
    payload = {"grant_type": "client_credentials", "expires_in": 1800}
    response = requests.post(
        "https://api.dolby.io/v1/auth/token",
        data=payload,
        auth=requests.auth.HTTPBasicAuth(APP_KEY, APP_SECRET),
    )
    return response.json()["access_token"]


def upload_media(file_path, headers):
    upload_url = "https://api.dolby.com/media/input"
    upload_body = {"url": f"dlb://in/{os.path.basename(file_path)}"}
    response = requests.post(upload_url, json=upload_body, headers=headers)
    response.raise_for_status()
    presigned_url = response.json()["url"]

    with open(file_path, "rb") as input_file:
        requests.put(presigned_url, data=input_file)


def create_enhancement_job(file_path, output_path, headers, audio_type):
    enhance_url = "https://api.dolby.com/media/enhance"
    enhance_body = {
        "input": f"dlb://in/{os.path.basename(file_path)}",
        "output": f"dlb://out/{os.path.basename(output_path)}",
        "content": {"type": audio_type},
    }
    response = requests.post(enhance_url, json=enhance_body, headers=headers)
    response.raise_for_status()
    return response.json()["job_id"]


def check_job_status(job_id, headers):
    status_url = "https://api.dolby.com/media/enhance"
    params = {"job_id": job_id}
    while True:
        response = requests.get(status_url, params=params, headers=headers)
        response.raise_for_status()
        status = response.json()["status"]
        if status == "Success":
            break
        print(f"Job status: {status}, progress: {response.json()['progress']}%")
        time.sleep(5)


def download_enhanced_file(output_path, headers):
    download_url = "https://api.dolby.com/media/output"
    args = {"url": f"dlb://out/{os.path.basename(output_path)}"}
    with requests.get(
        download_url, params=args, headers=headers, stream=True
    ) as response:
        response.raise_for_status()
        response.raw.decode_content = True
        print(f"Downloading from {response.url} into {output_path}")
        with open(output_path, "wb") as output_file:
            shutil.copyfileobj(response.raw, output_file)


def dolby_process(input_file, output_file, audio_type):
    access_token = get_access_token()
    headers = {"Authorization": f"Bearer {access_token}"}
    upload_media(input_file, headers)
    job_id = create_enhancement_job(input_file, output_file, headers, audio_type)
    check_job_status(job_id, headers)
    download_enhanced_file(output_file, headers)


def enhance_audio(recording, upload, audio_type):
    audio_type = audio_type_mapping[audio_type]
    if recording is not None:
        rate, data = recording
        temp_input_file = "input.wav"
    elif upload is not None:
        rate, data = upload
        if rate not in [44100, 48000] or data.dtype not in [np.int16, np.int32]:
            return None, None, "Invalid file type. Please upload an MP3 file."
        temp_input_file = "input.mp3"
    else:
        return (
            None,
            None,
            "Invalid input. Please record some audio or upload an audio file.",
        )

    write(temp_input_file, rate, data)

    temp_output_file = "output.wav"
    dolby_process(
        temp_input_file, temp_output_file, audio_type
    )  # Pass the audio type to the Dolby processing function

    return temp_input_file, temp_output_file, "Processing complete!"


def clone_voice(temp_output_file):
    # Your voice cloning logic goes here
    cloned_voice_file = "cloned_voice.wav"
    return cloned_voice_file, "Voice cloning complete!"


audio_type_mapping = {
    "Conference": "conference",
    "Interview": "interview",
    "Lecture": "lecture",
    "Meeting": "meeting",
    "Mobile Phone": "mobile_phone",
    "Music": "music",
    "Podcast": "podcast",
    "Studio": "studio",
    "Voice Over": "voice_over",
}

from gradio import Checkbox


def combined_function(
    recording, upload, audio_type, proceed_to_clone, name, description, labels, model
):
    input_file, output_file, status1 = enhance_audio(recording, upload, audio_type)
    status1 = "Enhancement complete!"
    transcript = transcribe_audio(output_file)
    if proceed_to_clone:
        clone_and_stream_voice(name, description, labels, transcript, model)
        status2 = "Cloning complete!"
    else:
        status2 = "Voice cloning not performed."
    return input_file, output_file, status1, transcript, status2


def main():
    iface = gr.Interface(
        fn=combined_function,
        inputs=[
            Audio(source="microphone", label="Recorded Audio"),
            Audio(source="upload", label="Uploaded Audio"),
            Radio(choices=list(audio_type_mapping.keys()), label="Audio Type"),
            Checkbox(label="Proceed to Clone Voice"),
            Textbox(label="Name"),
            Textbox(label="Description"),
            Textbox(label="Labels"),
            Radio(
                choices=["eleven_monolingual_v1", "eleven_multilingual_v1"],
                label="Model",
            ),
        ],
        outputs=[
            Audio(type="filepath", label="Original Audio"),
            Audio(type="filepath", label="Processed Audio"),
            Textbox(label="Enhancement Status"),
            Textbox(label="Transcript"),
            Textbox(label="Cloning Status"),
        ],
        title="Audio Enhancer, Transcriber and Voice Cloner",
        description="Enhance your audio, transcribe it and clone voices using the Dolby API",
        allow_flagging="never",

    )

    iface.launch(server_name="0.0.0.0", server_port=7860,share=True)


if __name__ == "__main__":

    main()