Update app.py

just copy from lingy

app.py

@@ -1,204 +1,242 @@
 import gradio as gr
-from gradio_leaderboard import Leaderboard, ColumnFilter, SelectColumns
-import pandas as pd
-from apscheduler.schedulers.background import BackgroundScheduler
-from huggingface_hub import snapshot_download
-
-from src.about import (
-    CITATION_BUTTON_LABEL,
-    CITATION_BUTTON_TEXT,
-    EVALUATION_QUEUE_TEXT,
-    INTRODUCTION_TEXT,
-    LLM_BENCHMARKS_TEXT,
-    TITLE,
-)
-from src.display.css_html_js import custom_css
-from src.display.utils import (
-    BENCHMARK_COLS,
-    COLS,
-    EVAL_COLS,
-    EVAL_TYPES,
-    AutoEvalColumn,
-    ModelType,
-    fields,
-    WeightType,
-    Precision
-)
-from src.envs import API, EVAL_REQUESTS_PATH, EVAL_RESULTS_PATH, QUEUE_REPO, REPO_ID, RESULTS_REPO, TOKEN
-from src.populate import get_evaluation_queue_df, get_leaderboard_df
-from src.submission.submit import add_new_eval
-
-
-def restart_space():
-    API.restart_space(repo_id=REPO_ID)
-
-### Space initialisation
-try:
-    print(EVAL_REQUESTS_PATH)
-    snapshot_download(
-        repo_id=QUEUE_REPO, local_dir=EVAL_REQUESTS_PATH, repo_type="dataset", tqdm_class=None, etag_timeout=30, token=TOKEN
+import requests
+import uuid
+import os
+from typing import Optional
+import tempfile
+from pydub import AudioSegment
+import re
+
+ASR_API = "http://astarwiz.com:9998/asr"
+TTS_SPEAK_SERVICE = 'http://astarwiz.com:9603/speak'
+TTS_WAVE_SERVICE = 'http://astarwiz.com:9603/wave'
+
+LANGUAGE_MAP = {
+    "en": "English",
+    "ma": "Malay",
+    "ta": "Tamil",
+    "zh": "Chinese"
+}
+
+# Add a password for developer mode
+DEVELOPER_PASSWORD = os.getenv("DEV_PWD")
+
+# Add this constant for the RapidAPI key
+RAPID_API_KEY = os.getenv("RAPID_API_KEY")
+
+# Add this constant for available speakers
+AVAILABLE_SPEAKERS = {
+    "en": ["MS"],
+    "ma": ["msFemale"],
+    "ta": ["ta_female1"],
+    "zh": ["childChinese2"]
+}
+
+def fetch_youtube_id(youtube_url: str) -> str:
+    if 'v=' in youtube_url:
+        return youtube_url.split("v=")[1].split("&")[0]
+    elif 'youtu.be/' in youtube_url:
+        return youtube_url.split("youtu.be/")[1]
+    elif 'shorts' in youtube_url:
+        return youtube_url.split("/")[-1]
+    else:
+        raise Exception("Unsupported URL format")
+
+def download_youtube_audio(youtube_url: str, output_dir: Optional[str] = None) -> Optional[str]:
+    video_id = fetch_youtube_id(youtube_url)
+    
+    if not video_id:
+        return None
+
+    if output_dir is None:
+        output_dir = tempfile.gettempdir()
+
+    output_filename = os.path.join(output_dir, f"{video_id}.mp3")
+    
+    if os.path.exists(output_filename):
+        return output_filename  # Return if the file already exists
+    
+    url = "https://youtube86.p.rapidapi.com/api/youtube/links"
+    headers = {
+        'Content-Type': 'application/json',
+        'x-rapidapi-host': 'youtube86.p.rapidapi.com',
+        'x-rapidapi-key': RAPID_API_KEY
+    }
+    data = {
+        "url": youtube_url
+    }
+    
+    response = requests.post(url, headers=headers, json=data)
+    print('Fetched audio links')
+    
+    if response.status_code == 200:
+        result = response.json()
+        for url in result[0]['urls']:
+            if url.get('isBundle'):
+                audio_url = url['url']
+                extension = url['extension']
+                audio_response = requests.get(audio_url)
+                
+                if audio_response.status_code == 200:
+                    temp_filename = os.path.join(output_dir, f"{video_id}.{extension}")
+                    with open(temp_filename, 'wb') as audio_file:
+                        audio_file.write(audio_response.content)
+                    
+                    # Convert to MP3 and downsample to 16000 Hz
+                    audio = AudioSegment.from_file(temp_filename, format=extension)
+                    audio = audio.set_frame_rate(16000)
+                    audio.export(output_filename, format="mp3", parameters=["-ar", "16000"])
+                    
+                    os.remove(temp_filename)  # Remove the temporary file
+                    return output_filename  # Return the final MP3 filename
+        
+        return None  # Return None if no successful download occurs
+    else:
+        print("Error:", response.status_code, response.text)
+        return None  # Return None on failure
+
+def inference_via_llm_api(input_text, min_new_tokens=2, max_new_tokens=64):
+    print(input_text)
+    one_vllm_input = f"<|im_start|>system\nYou are a translation expert.<|im_end|>\n<|im_start|>user\n{input_text}<|im_end|>\n<|im_start|>assistant"
+    vllm_api = 'http://astarwiz.com:2333/' + "v1/completions"
+    data = {
+        "prompt": one_vllm_input,
+        'model': "./Edu-4B-NewTok-V2-20240904/",
+        'min_tokens': min_new_tokens,
+        'max_tokens': max_new_tokens,
+        'temperature': 0.1,
+        'top_p': 0.75,
+        'repetition_penalty': 1.1,
+        "stop_token_ids": [151645, ],
+    }
+    response = requests.post(vllm_api, headers={"Content-Type": "application/json"}, json=data).json()
+    print(response)
+    if "choices" in response.keys():
+        return response["choices"][0]['text'].strip()
+    else:
+        return "The system got some error during vLLM generation. Please try it again."
+    
+def transcribe_and_speak(audio, source_lang, target_lang, youtube_url=None, target_speaker=None):
+    if youtube_url:
+        audio = download_youtube_audio(youtube_url)
+        if not audio:
+            return "Failed to download YouTube audio.", None, None
+
+    if not audio:
+        return "Please provide an audio input or a valid YouTube URL.", None, None
+
+    # ASR
+    file_id = str(uuid.uuid4())
+    files = {'file': open(audio, 'rb')}
+    data = {
+        'language': 'ms' if source_lang == 'ma' else source_lang,
+        'model_name': 'whisper-large-v2-local-cs',
+        'with_timestamp': False
+    }
+
+    asr_response = requests.post(ASR_API, files=files, data=data)
+    print(asr_response.json())
+    if asr_response.status_code == 200:
+        transcription = asr_response.json()['text']
+    else:
+        return "ASR failed", None, None
+
+    translation_prompt = f"Translate the following text from {LANGUAGE_MAP[source_lang]} to {LANGUAGE_MAP[target_lang]}: {transcription}"
+    translated_text = inference_via_llm_api(translation_prompt)
+    print(f"Translation: {translated_text}")
+
+    # TTS
+    tts_params = {
+        'language': target_lang,
+        'speed': 1.1,
+        'speaker': target_speaker or AVAILABLE_SPEAKERS[target_lang][0],  # Use the first speaker as default
+        'text': translated_text
+    }
+    
+    tts_response = requests.get(TTS_SPEAK_SERVICE, params=tts_params)
+    if tts_response.status_code == 200:
+        audio_file = tts_response.text.strip()
+        audio_url = f"{TTS_WAVE_SERVICE}?file={audio_file}"
+        return transcription, translated_text, audio_url
+    else:
+        return transcription, translated_text, "TTS failed"
+
+def check_password(password):
+    return password == DEVELOPER_PASSWORD
+
+def run_speech_translation(audio, source_lang, target_lang, youtube_url, target_speaker):
+    transcription, translated_text, audio_url = transcribe_and_speak(audio, source_lang, target_lang, youtube_url, target_speaker)
+    
+    return transcription, translated_text, audio_url
+
+with gr.Blocks() as demo:
+    gr.Markdown("# Speech Translation")
+    
+    # with gr.Tab("User Mode"):
+    gr.Markdown("Speak into the microphone, upload an audio file, or provide a YouTube URL. The app will translate and speak it back to you.")
+    
+    with gr.Row():
+        user_audio_input = gr.Audio(sources=["microphone", "upload"], type="filepath")
+        user_youtube_url = gr.Textbox(label="YouTube URL (optional)")
+    
+    with gr.Row():
+        user_source_lang = gr.Dropdown(choices=["en", "ma", "ta", "zh"], label="Source Language", value="en")
+        user_target_lang = gr.Dropdown(choices=["en", "ma", "ta", "zh"], label="Target Language", value="zh")
+        user_target_speaker = gr.Dropdown(choices=AVAILABLE_SPEAKERS['zh'], label="Target Speaker", value="childChinese2")
+
+    with gr.Row():
+        user_button = gr.Button("Translate and Speak", interactive=False)
+    
+    with gr.Row():
+        user_transcription_output = gr.Textbox(label="Transcription")
+        user_translation_output = gr.Textbox(label="Translation")
+        user_audio_output = gr.Audio(label="Translated Speech")
+    
+    user_video_output = gr.HTML(label="YouTube Video")
+
+    def update_button_state(audio, youtube_url):
+        print(audio, youtube_url)
+        return gr.Button(interactive=bool(audio) or bool(youtube_url))
+
+    user_audio_input.change(
+        fn=update_button_state,
+        inputs=[user_audio_input, user_youtube_url],
+        outputs=user_button
     )
-except Exception:
-    restart_space()
-try:
-    print(EVAL_RESULTS_PATH)
-    snapshot_download(
-        repo_id=RESULTS_REPO, local_dir=EVAL_RESULTS_PATH, repo_type="dataset", tqdm_class=None, etag_timeout=30, token=TOKEN
+    user_youtube_url.change(
+        fn=update_button_state,
+        inputs=[user_audio_input, user_youtube_url],
+        outputs=user_button
     )
-except Exception:
-    restart_space()
-
-
-LEADERBOARD_DF = get_leaderboard_df(EVAL_RESULTS_PATH, EVAL_REQUESTS_PATH, COLS, BENCHMARK_COLS)
-
-(
-    finished_eval_queue_df,
-    running_eval_queue_df,
-    pending_eval_queue_df,
-) = get_evaluation_queue_df(EVAL_REQUESTS_PATH, EVAL_COLS)
-
-def init_leaderboard(dataframe):
-    if dataframe is None or dataframe.empty:
-        raise ValueError("Leaderboard DataFrame is empty or None.")
-    return Leaderboard(
-        value=dataframe,
-        datatype=[c.type for c in fields(AutoEvalColumn)],
-        select_columns=SelectColumns(
-            default_selection=[c.name for c in fields(AutoEvalColumn) if c.displayed_by_default],
-            cant_deselect=[c.name for c in fields(AutoEvalColumn) if c.never_hidden],
-            label="Select Columns to Display:",
-        ),
-        search_columns=[AutoEvalColumn.model.name, AutoEvalColumn.license.name],
-        hide_columns=[c.name for c in fields(AutoEvalColumn) if c.hidden],
-        filter_columns=[
-            ColumnFilter(AutoEvalColumn.model_type.name, type="checkboxgroup", label="Model types"),
-            ColumnFilter(AutoEvalColumn.precision.name, type="checkboxgroup", label="Precision"),
-            ColumnFilter(
-                AutoEvalColumn.params.name,
-                type="slider",
-                min=0.01,
-                max=150,
-                label="Select the number of parameters (B)",
-            ),
-            ColumnFilter(
-                AutoEvalColumn.still_on_hub.name, type="boolean", label="Deleted/incomplete", default=True
-            ),
-        ],
-        bool_checkboxgroup_label="Hide models",
-        interactive=False,
+    
+    user_button.click(
+        fn=run_speech_translation,
+        inputs=[user_audio_input, user_source_lang, user_target_lang, user_youtube_url, user_target_speaker],
+        outputs=[user_transcription_output, user_translation_output, user_audio_output]
     )
 
+    def update_video_embed(youtube_url):
+        if youtube_url:
+            try:
+                video_id = fetch_youtube_id(youtube_url)
+                return f'<iframe width="560" height="315" src="https://www.youtube.com/embed/{video_id}" frameborder="0" allow="autoplay; encrypted-media" allowfullscreen></iframe>'
+            except Exception as e:
+                print(f"Error embedding video: {e}")
+        return ""
+
+    user_youtube_url.change(
+        fn=update_video_embed,
+        inputs=[user_youtube_url],
+        outputs=[user_video_output]
+    )
 
-demo = gr.Blocks(css=custom_css)
-with demo:
-    gr.HTML(TITLE)
-    gr.Markdown(INTRODUCTION_TEXT, elem_classes="markdown-text")
-
-    with gr.Tabs(elem_classes="tab-buttons") as tabs:
-        with gr.TabItem("🏅 LLM Benchmark", elem_id="llm-benchmark-tab-table", id=0):
-            leaderboard = init_leaderboard(LEADERBOARD_DF)
-
-        with gr.TabItem("📝 About", elem_id="llm-benchmark-tab-table", id=2):
-            gr.Markdown(LLM_BENCHMARKS_TEXT, elem_classes="markdown-text")
-
-        with gr.TabItem("🚀 Submit here! ", elem_id="llm-benchmark-tab-table", id=3):
-            with gr.Column():
-                with gr.Row():
-                    gr.Markdown(EVALUATION_QUEUE_TEXT, elem_classes="markdown-text")
-
-                with gr.Column():
-                    with gr.Accordion(
-                        f"✅ Finished Evaluations ({len(finished_eval_queue_df)})",
-                        open=False,
-                    ):
-                        with gr.Row():
-                            finished_eval_table = gr.components.Dataframe(
-                                value=finished_eval_queue_df,
-                                headers=EVAL_COLS,
-                                datatype=EVAL_TYPES,
-                                row_count=5,
-                            )
-                    with gr.Accordion(
-                        f"🔄 Running Evaluation Queue ({len(running_eval_queue_df)})",
-                        open=False,
-                    ):
-                        with gr.Row():
-                            running_eval_table = gr.components.Dataframe(
-                                value=running_eval_queue_df,
-                                headers=EVAL_COLS,
-                                datatype=EVAL_TYPES,
-                                row_count=5,
-                            )
-
-                    with gr.Accordion(
-                        f"⏳ Pending Evaluation Queue ({len(pending_eval_queue_df)})",
-                        open=False,
-                    ):
-                        with gr.Row():
-                            pending_eval_table = gr.components.Dataframe(
-                                value=pending_eval_queue_df,
-                                headers=EVAL_COLS,
-                                datatype=EVAL_TYPES,
-                                row_count=5,
-                            )
-            with gr.Row():
-                gr.Markdown("# ✉️✨ Submit your model here!", elem_classes="markdown-text")
-
-            with gr.Row():
-                with gr.Column():
-                    model_name_textbox = gr.Textbox(label="Model name")
-                    revision_name_textbox = gr.Textbox(label="Revision commit", placeholder="main")
-                    model_type = gr.Dropdown(
-                        choices=[t.to_str(" : ") for t in ModelType if t != ModelType.Unknown],
-                        label="Model type",
-                        multiselect=False,
-                        value=None,
-                        interactive=True,
-                    )
-
-                with gr.Column():
-                    precision = gr.Dropdown(
-                        choices=[i.value.name for i in Precision if i != Precision.Unknown],
-                        label="Precision",
-                        multiselect=False,
-                        value="float16",
-                        interactive=True,
-                    )
-                    weight_type = gr.Dropdown(
-                        choices=[i.value.name for i in WeightType],
-                        label="Weights type",
-                        multiselect=False,
-                        value="Original",
-                        interactive=True,
-                    )
-                    base_model_name_textbox = gr.Textbox(label="Base model (for delta or adapter weights)")
-
-            submit_button = gr.Button("Submit Eval")
-            submission_result = gr.Markdown()
-            submit_button.click(
-                add_new_eval,
-                [
-                    model_name_textbox,
-                    base_model_name_textbox,
-                    revision_name_textbox,
-                    precision,
-                    weight_type,
-                    model_type,
-                ],
-                submission_result,
-            )
+    def update_target_speakers(target_lang):
+        return gr.Dropdown(choices=AVAILABLE_SPEAKERS[target_lang], value=AVAILABLE_SPEAKERS[target_lang][0])
 
-    with gr.Row():
-        with gr.Accordion("📙 Citation", open=False):
-            citation_button = gr.Textbox(
-                value=CITATION_BUTTON_TEXT,
-                label=CITATION_BUTTON_LABEL,
-                lines=20,
-                elem_id="citation-button",
-                show_copy_button=True,
-            )
-
-scheduler = BackgroundScheduler()
-scheduler.add_job(restart_space, "interval", seconds=1800)
-scheduler.start()
-demo.queue(default_concurrency_limit=40).launch()
+    user_target_lang.change(
+        fn=update_target_speakers,
+        inputs=[user_target_lang],
+        outputs=[user_target_speaker]
+    )
+
+demo.launch(auth=("test", "test")