app.py

import logging
import sys
import gradio as gr
import vosk
import json
import subprocess

logging.basicConfig(
    format="%(asctime)s - %(levelname)s - %(name)s -   %(message)s",
    datefmt="%m/%d/%Y %H:%M:%S",
    handlers=[logging.StreamHandler(sys.stdout)],
)
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)

LARGE_MODEL_BY_LANGUAGE = {
    "Russian": {"model_id": "vosk-model-ru-0.42"},
    "Chinese": {"model_id": "vosk-model-cn-0.22"},
    "English": {"model_id": "vosk-model-en-us-0.22"},
    "French": {"model_id": "vosk-model-fr-0.22"},
    "German": {"model_id": "vosk-model-de-0.22"},
    "Italian": {"model_id": "vosk-model-it-0.22"},
    "Japanese": {"model_id": "vosk-model-ja-0.22"},
    "Hindi": {"model_id": "vosk-model-hi-0.22"},
    "Persian": {"model_id": "vosk-model-fa-0.5"},
    "Uzbek": {"model_id": "vosk-model-small-uz-0.22"},
}

LANGUAGES = sorted(LARGE_MODEL_BY_LANGUAGE.keys())
CACHED_MODELS_BY_ID = {}

def asr(model, input_file):

    rec = vosk.KaldiRecognizer(model, 16000.0)
    results = []

    process = subprocess.Popen(f'ffmpeg -loglevel quiet -i {input_file} -ar 16000 -ac 1 -f s16le -'.split(),
                            stdout=subprocess.PIPE)

    while True:
        data = process.stdout.read(4000)
        if len(data) == 0:
            break
        if rec.AcceptWaveform(data):
            jres = json.loads(rec.Result())
            results.append(jres['text'])

    jres = json.loads(rec.FinalResult())
    results.append(jres['text'])

    return " ".join(results)


def run(input_file, language, history):

    logger.info(f"Running ASR for {language} for {input_file}")

    history = history or []

    model = LARGE_MODEL_BY_LANGUAGE.get(language, None)

    if model is None:
        history.append({
            "error_message": f"Failed to find a model for {language} language :("
        })
    elif input_file is None:
        history.append({
            "error_message": f"Record input audio first"
        })
    else:
        model_instance = CACHED_MODELS_BY_ID.get(model["model_id"], None)
        if model_instance is None:
            model_instance = vosk.Model(model_name=model["model_id"])
            CACHED_MODELS_BY_ID[model["model_id"]] = model_instance

        transcription = asr(model_instance, input_file)

        logger.info(f"Transcription for {input_file}: {transcription}")

        history.append({
            "model_id": model["model_id"],
            "language": language,
            "transcription": transcription,
            "error_message": None
        })

    html_output = "<div class='result'>"
    for item in history:
        if item["error_message"] is not None:
            html_output += f"<div class='result_item result_item_error'>{item['error_message']}</div>"
        else:
            html_output += "<div class='result_item result_item_success'>"
            html_output += f'{item["transcription"]}<br/>'
            html_output += "</div>"
    html_output += "</div>"

    return html_output, history


gr.Interface(
    run,
    inputs=[
        gr.inputs.Audio(source="microphone", type="filepath", label="Record something..."),
        gr.inputs.Radio(label="Language", choices=LANGUAGES),
        "state"
    ],
    outputs=[
        gr.outputs.HTML(label="Outputs"),
        "state"
    ],
    title="Automatic Speech Recognition",
    description="",
    css="""
    .result {display:flex;flex-direction:column}
    .result_item {padding:15px;margin-bottom:8px;border-radius:15px;width:100%}
    .result_item_success {background-color:mediumaquamarine;color:white;align-self:start}
    .result_item_error {background-color:#ff7070;color:white;align-self:start}
    """,
    allow_flagging="never",
    theme="default"
).launch(enable_queue=True)