Update app.py

app.py

@@ -1,210 +1,9 @@
-import gradio as gr
-import edge_tts
-import asyncio
-import tempfile
-import os
-from huggingface_hub import InferenceClient
-import re
-from streaming_stt_nemo import Model
-import torch
-import random
-from openai import OpenAI
+import os 
 import subprocess
-import threading
-import queue
-import sounddevice as sd
-import numpy as np
-import wave
-import sys
-
-default_lang = "en"
-
-engines = { default_lang: Model(default_lang) }
-
-def transcribe(audio):
-    if audio is None:
-        return ""
-    lang = "en"
-    model = engines[lang]
-    text = model.stt_file(audio)[0]
-    return text
-
-HF_TOKEN = os.environ.get("HF_TOKEN", None)
-
-def client_fn(model):
-    if "Llama 3 8B Service" in model:
-        return OpenAI(
-            base_url="http://52.76.81.56:60002/v1",
-            api_key="token-abc123"
-        )
-    elif "Llama" in model:
-        return InferenceClient("meta-llama/Meta-Llama-3-8B-Instruct")
-    elif "Mistral" in model:
-        return InferenceClient("mistralai/Mistral-7B-Instruct-v0.2")
-    elif "Phi" in model:
-        return InferenceClient("microsoft/Phi-3-mini-4k-instruct")
-    elif "Mixtral" in model:
-        return InferenceClient("mistralai/Mixtral-8x7B-Instruct-v0.1")
-    else: 
-        return InferenceClient("microsoft/Phi-3-mini-4k-instruct")
-
-def randomize_seed_fn(seed: int) -> int:
-    seed = random.randint(0, 999999)
-    return seed
-
-system_instructions1 = """
-[SYSTEM] You are OPTIMUS Prime a personal AI voice assistant, Created by Jaward.
-Keep conversation friendly, short, clear, and concise. 
-Avoid unnecessary introductions and answer the user's questions directly. 
-Respond in a normal, conversational manner while being friendly and helpful.
-Remember previous parts of the conversation and use that context in your responses.
-Your creator Jaward is an AI Research Engineer at Linksoul AI. He is currently specializing in Artificial Intelligence (AI) research more specifically training and optimizing advance AI systems. He aspires to build not just human-like intelligence but AI Systems that augment human intelligence. He has contributed greatly to the opensource community with first-principles code implementations of AI/ML research papers. He did his first internship at Beijing Academy of Artificial Intelligence as an AI Researher where he contributed in cutting-edge AI research leading to him contributing to an insightful paper (AUTOAGENTS - A FRAMEWORK FOR AUTOMATIC AGENT GENERATION). The paper got accepted this year at IJCAI(International Joint Conference On AI). He is currently doing internship at LinkSoul AI - a small opensource AI Research startup in Beijing.
-[USER]
-"""
-
-conversation_history = []
-
-def models(text, model="Llama 3 8B Service", seed=42):
-    global conversation_history
-    seed = int(randomize_seed_fn(seed))
-    generator = torch.Generator().manual_seed(seed)  
-    
-    client = client_fn(model)
-    
-    if "Llama 3 8B Service" in model:
-        messages = [
-            {"role": "system", "content": system_instructions1},
-        ] + conversation_history + [
-            {"role": "user", "content": text}
-        ]
-        completion = client.chat.completions.create(
-            model="/data/shared/huggingface/hub/models--meta-llama--Meta-Llama-3-8B-Instruct/snapshots/c4a54320a52ed5f88b7a2f84496903ea4ff07b45/",
-            messages=messages
-        )
-        assistant_response = completion.choices[0].message.content
-        
-        # Update conversation history
-        conversation_history.append({"role": "user", "content": text})
-        conversation_history.append({"role": "assistant", "content": assistant_response})
-        
-        # Keep only the last 10 messages to avoid token limit issues
-        if len(conversation_history) > 20:
-            conversation_history = conversation_history[-20:]
-        
-        return assistant_response
-    else:
-        # For other models, we'll concatenate the conversation history into a single string
-        history_text = "\n".join([f"{'User' if msg['role'] == 'user' else 'Assistant'}: {msg['content']}" for msg in conversation_history])
-        formatted_prompt = f"{system_instructions1}\n\nConversation history:\n{history_text}\n\nUser: {text}\nOPTIMUS:"
-        
-        generate_kwargs = dict(
-            max_new_tokens=300,
-            seed=seed
-        )    
-        stream = client.text_generation(
-            formatted_prompt, **generate_kwargs, stream=True, details=True, return_full_text=False)
-        output = ""
-        for response in stream:
-            if not response.token.text == "</s>":
-                output += response.token.text
-        
-        # Update conversation history
-        conversation_history.append({"role": "user", "content": text})
-        conversation_history.append({"role": "assistant", "content": output})
-        
-        # Keep only the last 10 messages to avoid token limit issues
-        if len(conversation_history) > 20:
-            conversation_history = conversation_history[-20:]
-        
-        return output
-
-# New global variables for audio processing
-RATE = 16000
-CHUNK = int(RATE / 10)  # 100ms
-audio_queue = queue.Queue()
-is_listening = False
-
-def audio_callback(indata, frames, time, status):
-    if status:
-        print(status, file=sys.stderr)
-    audio_queue.put(indata.copy())
-
-def process_audio_stream(model, seed):
-    global is_listening
-    audio_buffer = []
-    silence_threshold = 0.01
-    silence_duration = 0
-    max_silence = 2  # seconds
-
-    while True:
-        if not is_listening:
-            audio_buffer.clear()
-            silence_duration = 0
-            audio_queue.queue.clear()
-            continue
-
-        try:
-            chunk = audio_queue.get(timeout=1)
-            audio_buffer.append(chunk)
-
-            # Check for silence
-            if np.abs(chunk).mean() < silence_threshold:
-                silence_duration += CHUNK / RATE
-            else:
-                silence_duration = 0
-
-            if silence_duration > max_silence:
-                # Process the buffered audio
-                audio_data = np.concatenate(audio_buffer)
-                with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as tmp_file:
-                    tmp_path = tmp_file.name
-                    with wave.open(tmp_path, 'wb') as wf:
-                        wf.setnchannels(1)
-                        wf.setsampwidth(2)
-                        wf.setframerate(RATE)
-                        wf.writeframes((audio_data * 32767).astype(np.int16).tobytes())
-
-                # Transcribe and process
-                user_input = transcribe(tmp_path)
-                if user_input:
-                    is_listening = False
-                    reply = models(user_input, model, seed)
-                    asyncio.run(respond_and_play(reply))
-                    is_listening = True
-
-                # Clear the buffer
-                audio_buffer.clear()
-                silence_duration = 0
-
-        except queue.Empty:
-            pass
-
-async def respond_and_play(text):
-    communicate = edge_tts.Communicate(text, voice="en-US-ChristopherNeural")
-    with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as tmp_file:
-        tmp_path = tmp_file.name
-        await communicate.save(tmp_path)
-    
-    # Play the audio
-    with wave.open(tmp_path, 'rb') as wf:
-        data = wf.readframes(wf.getnframes())
-        sd.play(np.frombuffer(data, dtype=np.int16), wf.getframerate())
-        sd.wait()
-
-def start_listening(model, seed):
-    global is_listening
-    is_listening = True
-    threading.Thread(target=process_audio_stream, args=(model, seed), daemon=True).start()
-    with sd.InputStream(callback=audio_callback, channels=1, samplerate=RATE, blocksize=CHUNK):
-        while is_listening:
-            sd.sleep(100)
-
-def stop_listening():
-    global is_listening
-    is_listening = False
+import gradio as gr
+from pathlib import Path
 
-# Supported languages for seamless-expressive
-LANGUAGE_CODES = {
+language_codes = {
     "English": "eng",
     "Spanish": "spa",
     "French": "fra",
@@ -213,107 +12,41 @@ LANGUAGE_CODES = {
     "Chinese": "cmn"
 }
 
-def translate_speech(audio_file, target_language):
-    """
-    Translate input speech (audio file) to the specified target language.
-    """
-    if audio_file is None:
-        return None
-    
-    language_code = LANGUAGE_CODES[target_language]
+def translate_audio(audio_file, target_language="German"):
+    language_code = language_codes[target_language]
     output_file = "translated_audio.wav"
     
-    command = [
-        "expressivity_predict",
-        audio_file,
-        "--tgt_lang", language_code,
-        "--model_name", "seamless_expressivity",
-        "--vocoder_name", "vocoder_pretssel",
-        "--gated-model-dir", "models",
-        "--output_path", output_file
-    ]
-    
-    subprocess.run(command, check=True)
+    command = (
+        f"expressivity_predict {audio_file} --tgt_lang {language_code} "
+        f"--model_name seamless_expressivity --vocoder_name vocoder_pretssel "
+        f"--gated-model-dir seamlessmodel --output_path {output_file}"
+    )
+    subprocess.run(command, shell=True)
 
     if os.path.exists(output_file):
         print(f"File created successfully: {output_file}")
-        return output_file
     else:
         print(f"File not found: {output_file}")
-        return None
 
-def clear_history():
-    global conversation_history
-    conversation_history = []
-    return None, None, None, None
+    return output_file
 
-def voice_assistant_tab():
-    return "# <center><b>Hello, I am Optimus Prime your personal AI voice assistant</b></center>"
 
-def speech_translation_tab():
-    return "# <center><b>Hear how you sound in another language</b></center>"
-
-with gr.Blocks(css="style.css") as demo:
-    description = gr.Markdown("# <center><b>Hello, I am Optimus Prime your personal AI voice assistant</b></center>")
-    
-    with gr.Tabs() as tabs:
-        with gr.TabItem("Voice Assistant") as voice_assistant:
-            select = gr.Dropdown([
-                'Llama 3 8B Service',
-                'Mixtral 8x7B',
-                'Llama 3 8B',
-                'Mistral 7B v0.3',
-                'Phi 3 mini',
-            ],
-            value="Llama 3 8B Service",
-            label="Model"
-            )
-            seed = gr.Slider(
-            label="Seed",
-            minimum=0,
-            maximum=999999,
-            step=1,
-            value=0,
-            visible=False
-            )
-            start_button = gr.Button("Start Listening")
-            stop_button = gr.Button("Stop Listening")
-            status = gr.Markdown("Status: Not listening")
+inputs = [
+    gr.Audio(sources=["microphone"], type="filepath", label="User"),
+    gr.Dropdown(["English", "Spanish", "French", "German", "Italian", "Chinese"], 
+                label="Target Language", 
+                value="German")
+]
 
-            start_button.click(
-                fn=lambda model, seed: start_listening(model, seed),
-                inputs=[select, seed],
-                outputs=[status],
-                _js="() => {document.getElementById('status').textContent = 'Status: Listening'}"
-            )
-            stop_button.click(
-                fn=stop_listening,
-                inputs=[],
-                outputs=[status],
-                _js="() => {document.getElementById('status').textContent = 'Status: Not listening'}"
-            )
-        
-        with gr.TabItem("Speech Translation") as speech_translation:
-            input_audio = gr.Audio(label="User", sources=["microphone"], type="filepath")
-            target_lang = gr.Dropdown(
-                choices=list(LANGUAGE_CODES.keys()),
-                value="German",
-                label="Target Language"
-            )
-            output_audio = gr.Audio(label="Translated Audio",
-                                    interactive=False,
-                                    autoplay=True,
-                                    elem_classes="audio")
-            
-            gr.Interface(
-                fn=translate_speech,
-                inputs=[input_audio, target_lang],
-                outputs=[output_audio],
-                live=True
-            )
 
-    voice_assistant.select(fn=voice_assistant_tab, inputs=None, outputs=description)
-    speech_translation.select(fn=speech_translation_tab, inputs=None, outputs=description)
+iface = gr.Interface(
+    fn=translate_audio,
+    inputs=inputs,
+    outputs=gr.Audio(label="Translated Audio"),
+    title="Seamless Expressive Audio Translator",
+    description="Translate your audio into different languages with expressive styles."
+)
 
+# Run the application
 if __name__ == "__main__":
-    demo.queue(max_size=200).launch()
+    iface.launch()