Update app.py

app.py

@@ -1,273 +1,135 @@
+import os
+import re
 import warnings
 import gradio as gr
-from transformers import pipeline
-from transformers import AutoProcessor
+from transformers import pipeline, AutoProcessor
 from pyctcdecode import build_ctcdecoder
 from transformers import Wav2Vec2ProcessorWithLM
+from indictrans import Transliterator
 
-import os
-import re
-#import torchaudio
-
-# Initialize the speech recognition pipeline and transliterator
-p1 = pipeline(task="automatic-speech-recognition", model="cdactvm/w2v-bert-odia_v1")
-p2 = pipeline(task="automatic-speech-recognition", model="cdactvm/w2v-bert-2.0-hindi_v1")
-#p3 = pipeline(task="automatic-speech-recognition", model="cdactvm/kannada_w2v-bert_model")
-#p4 = pipeline(task="automatic-speech-recognition", model="cdactvm/telugu_w2v-bert_model")
-#p5 = pipeline(task="automatic-speech-recognition", model="Sajjo/w2v-bert-2.0-bangala-gpu-CV16.0_v2")
-#p6 = pipeline(task="automatic-speech-recognition", model="cdactvm/hf-open-assames")
-p7 = pipeline(task="automatic-speech-recognition", model="cdactvm/w2v-assames")
+# Initialize ASR pipelines
+asr_models = {
+    "Odiya": pipeline(task="automatic-speech-recognition", model="cdactvm/w2v-bert-2.0-odia_v1"),
+    "Odiya-trans": pipeline(task="automatic-speech-recognition", model="cdactvm/w2v-bert-2.0-odia_v1"),
+    "Hindi": pipeline(task="automatic-speech-recognition", model="cdactvm/w2v-bert-2.0-hindi_v1"),
+    "Hindi-trans": pipeline(task="automatic-speech-recognition", model="cdactvm/w2v-bert-2.0-hindi_v1"),
+    # Add other models similarly
+    # "Kannada": pipeline(...),
+    # "Telugu": pipeline(...),
+    # "Bangala": pipeline(...),
+    "Assamese-Model2": pipeline(task="automatic-speech-recognition", model="cdactvm/w2v-assames"),
+}
+
+# Initialize Assamese model with Language Model
 processor = AutoProcessor.from_pretrained("cdactvm/w2v-assames")
 vocab_dict = processor.tokenizer.get_vocab()
 sorted_vocab_dict = {k.lower(): v for k, v in sorted(vocab_dict.items(), key=lambda item: item[1])}
-decoder = build_ctcdecoder(
-    labels=list(sorted_vocab_dict.keys()),
-    kenlm_model_path="lm.binary",
-    )
-processor_with_lm = Wav2Vec2ProcessorWithLM(
-    feature_extractor=processor.feature_extractor,
-    tokenizer=processor.tokenizer,
-    decoder=decoder
-    )
-processor.feature_extractor._processor_class = "Wav2Vec2ProcessorWithLM"
-p8 = pipeline("automatic-speech-recognition", model="cdactvm/w2v-assames", tokenizer=processor_with_lm, feature_extractor=processor_with_lm.feature_extractor, decoder=processor_with_lm.decoder)
-    
-
-os.system('git clone https://github.com/irshadbhat/indic-trans.git')
-os.system('pip install ./indic-trans/.')
-
-#HF_TOKEN = os.getenv('HF_TOKEN')
-#hf_writer = gr.HuggingFaceDatasetSaver(HF_TOKEN, "asr_demo")
-
-from indictrans import Transliterator
-trn = Transliterator(source='ori', target='eng', build_lookup=True)
-def transcribe_odiya(speech):
-    text = p1(speech)["text"]
-    if text is None:
-        return "Error: ASR returned None"
-    return text
-
-
+decoder = build_ctcdecoder(labels=list(sorted_vocab_dict.keys()), kenlm_model_path="lm.binary")
+processor_with_lm = Wav2Vec2ProcessorWithLM(feature_extractor=processor.feature_extractor,
+                                            tokenizer=processor.tokenizer,
+                                            decoder=decoder)
+asr_models["Assamese-LM"] = pipeline("automatic-speech-recognition", model="cdactvm/w2v-assames",
+                                     tokenizer=processor_with_lm,
+                                     feature_extractor=processor_with_lm.feature_extractor,
+                                     decoder=processor_with_lm.decoder)
+
+# Initialize Transliterator
+transliterators = {
+    "Odiya-trans": Transliterator(source='ori', target='eng', build_lookup=True),
+    "Hindi-trans": Transliterator(source='hin', target='eng', build_lookup=True),
+    # Add other transliterators similarly
+}
+
+# Function to clean HTML tags from text
 def cleanhtml(raw_html):
-    cleantext = re.sub(r'<.*?>', '', raw_html)
-    return cleantext
+    return re.sub(r'<.*?>', '', raw_html)
 
-def transcribe_hindi(speech):
-    text = p2(speech)["text"]
-    if text is None:
-        return "Error: ASR returned None"
-    return text
-def transcribe_kannada(speech):
-    text = p3(speech)["text"]
-    if text is None:
-        return "Error: ASR returned None"
-    return text
-def transcribe_telugu(speech):
-    text = p4(speech)["text"]
-    if text is None:
-        return "Error: ASR returned None"
-    return text
-    
-def transcribe_bangala(speech):
-    text = p5(speech)["text"]
-    if text is None:
-        return "Error: ASR returned None"
-    return text
+# Transcribe audio using the selected model
+def transcribe(lng, speech, transliterate=False):
+    model = asr_models.get(lng)
+    if not model:
+        return f"Unsupported language: {lng}"
 
-def transcribe_assamese_LM(speech):
-    text = p8(speech)["text"]
-    text = cleanhtml(text)
+    result = model(speech)
+    text = result.get("text")
     if text is None:
         return "Error: ASR returned None"
-    return text
-    
-def transcribe_assamese_model2(speech):
-    text = p7(speech)["text"]
-    text = cleanhtml(text)
-    if text is None:
-        return "Error: ASR returned None"
-    return text
-    
-def transcribe_odiya_eng(speech):
-    trn = Transliterator(source='ori', target='eng', build_lookup=True)
-    text = p1(speech)["text"]
-    if text is None:
-        return "Error: ASR returned None"
-    sentence = trn.transform(text)
-    if sentence is None:
-        return "Error: Transliteration returned None"
-    replaced_words = replace_words(sentence)
-    processed_sentence = process_doubles(replaced_words)
-    return process_transcription(processed_sentence)
-    
-def transcribe_ban_eng(speech):
-    trn = Transliterator(source='ben', target='eng', build_lookup=True)
-    text = p5(speech)["text"]
-    if text is None:
-        return "Error: ASR returned None"
-    sentence = trn.transform(text)
-    if sentence is None:
-        return "Error: Transliteration returned None"
-    replaced_words = replace_words(sentence)
-    processed_sentence = process_doubles(replaced_words)
-    return process_transcription(processed_sentence)
 
-def transcribe_hin_eng(speech):
-    trn = Transliterator(source='hin', target='eng', build_lookup=True)
-    text = p2(speech)["text"]
-    if text is None:
-        return "Error: ASR returned None"
-    sentence = trn.transform(text)
-    if sentence is None:
-        return "Error: Transliteration returned None"
-    replaced_words = replace_words(sentence)
-    processed_sentence = process_doubles(replaced_words)
-    return process_transcription(processed_sentence)
+    if transliterate:
+        trn = transliterators.get(lng + "-trans")
+        if not trn:
+            return f"Transliterator not available for: {lng}"
 
-def transcribe_kan_eng(speech):
-    trn = Transliterator(source='kan', target='eng', build_lookup=True)
-    text = p3(speech)["text"]
-    if text is None:
-        return "Error: ASR returned None"
-    sentence = trn.transform(text)
-    if sentence is None:
-        return "Error: Transliteration returned None"
-    replaced_words = replace_words(sentence)
-    processed_sentence = process_doubles(replaced_words)
-    return process_transcription(processed_sentence)
-    
-def transcribe_tel_eng(speech):
-    trn = Transliterator(source='tel', target='eng', build_lookup=True)
-    text = p4(speech)["text"]
-    if text is None:
-        return "Error: ASR returned None"
-    sentence = trn.transform(text)
-    if sentence is None:
-        return "Error: Transliteration returned None"
-    replaced_words = replace_words(sentence)
-    processed_sentence = process_doubles(replaced_words)
-    return process_transcription(processed_sentence)
-
-
-def process_transcription(input_sentence):
-    word_to_code_map = {}
-    code_to_word_map = {}
+        sentence = trn.transform(text)
+        if sentence is None:
+            return "Error: Transliteration returned None"
+        return process_transcription(sentence)
 
-    transcript_1 = sentence_to_transcript(input_sentence, word_to_code_map)
-    if transcript_1 is None:
-        return "Error: Transcript conversion returned None"
+    return cleanhtml(text)
 
-    numbers = text2int(transcript_1)
-    if numbers is None:
-        return "Error: Text to number conversion returned None"
+# Function to process and correct transcriptions
+def process_transcription(sentence):
+    replaced_words = replace_words(sentence)
+    processed_sentence = process_doubles(replaced_words)
+    return convert_to_text(processed_sentence)
 
-    code_to_word_map = {v: k for k, v in word_to_code_map.items()}
-    text = transcript_to_sentence(numbers, code_to_word_map)
-    return text
-
-def sel_lng(lng, mic=None, file=None):
-    if mic is not None:
-        audio = mic
-    elif file is not None:
-        audio = file
-    else:
-        return "You must either provide a mic recording or a file"
-    
-    if lng == "Odiya":
-        return transcribe_odiya(audio)
-    elif lng == "Odiya-trans":
-        return transcribe_odiya_eng(audio)
-    elif lng == "Hindi-trans":
-        return transcribe_hin_eng(audio)
-    elif lng == "Hindi":
-        return transcribe_hindi(audio)
-    elif lng == "Kannada-trans":
-       return transcribe_kan_eng(audio)
-    elif lng == "Kannada":
-       return transcribe_kannada(audio)
-    elif lng == "Telugu-trans":
-       return transcribe_tel_eng(audio)
-    elif lng == "Telugu":
-       return transcribe_telugu(audio)
-    elif lng == "Bangala-trans":
-       return transcribe_ban_eng(audio)
-    elif lng == "Bangala":
-       return transcribe_bangala(audio)
-    elif lng == "Assamese-LM":
-       return transcribe_assamese_LM(audio)
-    elif lng == "Assamese-Model2":
-       return transcribe_assamese_model2(audio)
-        
-    
-# Function to replace incorrectly spelled words
+# Replace incorrectly spelled words
 def replace_words(sentence):
     replacements = [
-        (r'\bjiro\b', 'zero'), (r'\bjero\b', 'zero'), (r'\bnn\b', 'one'),
-        (r'\bn\b', 'one'), (r'\bna\b', 'one'), (r'\btu\b', 'two'),
-        (r'\btoo\b', 'two'), (r'\bthiri\b', 'three'), (r'\bfor\b', 'four'),
-        (r'\bfore\b', 'four'), (r'\bfib\b', 'five'), (r'\bdublseven\b', 'double seven'),
-        (r'\bdubalathri\b', 'double three'), (r'\bnineeit\b', 'nine eight'),
+        (r'\bjiro\b', 'zero'), (r'\bjero\b', 'zero'),
+        (r'\bnn\b', 'one'), (r'\bn\b', 'one'), (r'\bvan\b', 'one'), (r'\bna\b', 'one'), (r'\bek\b', 'one'),
+        (r'\btu\b', 'two'), (r'\btoo\b', 'two'), (r'\bdo\b', 'two'),
+        (r'\bthiri\b', 'three'), (r'\btiri\b', 'three'), (r'\bdubalathri\b', 'double three'), (r'\btin\b', 'three'),
+        (r'\bfor\b', 'four'), (r'\bfore\b', 'four'),
+        (r'\bfib\b', 'five'), (r'\bpaanch\b', 'five'),
+        (r'\bchha\b', 'six'), (r'\bchhah\b', 'six'), (r'\bchau\b', 'six'),
+        (r'\bdublseven\b', 'double seven'), (r'\bsath\b', 'seven'),
+        (r'\baath\b', 'eight'),
+        (r'\bnau\b', 'nine'),
+        (r'\bdas\b', 'ten'),
+        (r'\bnineeit\b', 'nine eight'),
         (r'\bfipeit\b', 'five eight'), (r'\bdubal\b', 'double'), (r'\bsevenatu\b', 'seven two'),
     ]
     for pattern, replacement in replacements:
         sentence = re.sub(pattern, replacement, sentence)
     return sentence
 
-# Function to process "double" followed by a number
+# Process "double" followed by a number
 def process_doubles(sentence):
     tokens = sentence.split()
     result = []
     i = 0
     while i < len(tokens):
-        if tokens[i] in ("double", "dubal"):
-            if i + 1 < len(tokens):
-                result.append(tokens[i + 1])
-                result.append(tokens[i + 1])
-                i += 2
-            else:
-                result.append(tokens[i])
-                i += 1
+        if tokens[i] in ("double", "dubal") and i + 1 < len(tokens):
+            result.extend([tokens[i + 1]] * 2)
+            i += 2
         else:
             result.append(tokens[i])
             i += 1
     return ' '.join(result)
 
-# Function to generate Soundex code for a word
-def soundex(word):
-    word = word.upper()
-    word = ''.join(filter(str.isalpha, word))
-    if not word:
-        return None
-    soundex_mapping = {
-        'B': '1', 'F': '1', 'P': '1', 'V': '1',
-        'C': '2', 'G': '2', 'J': '2', 'K': '2', 'Q': '2', 'S': '2', 'X': '2', 'Z': '2',
-        'D': '3', 'T': '3', 'L': '4', 'M': '5', 'N': '5', 'R': '6'
-    }
-    soundex_code = word[0]
-    for char in word[1:]:
-        if char not in ('H', 'W'):
-            soundex_code += soundex_mapping.get(char, '0')
-            soundex_code = soundex_code[0] + ''.join(c for i, c in enumerate(soundex_code[1:]) if c != soundex_code[i])
-            soundex_code = soundex_code.replace('0', '') + '000'
-    return soundex_code[:4]
-
-# Function to convert text to numerical representation
-def is_number(x):
-    if type(x) == str:
-        x = x.replace(',', '')
-    try:
-        float(x)
-    except:
-        return False
-    return True
+# Convert Soundex code back to text
+def convert_to_text(input_sentence):
+    word_to_code_map = {}
+    transcript = sentence_to_transcript(input_sentence, word_to_code_map)
+    if transcript is None:
+        return "Error: Transcript conversion returned None"
+
+    numbers = text2int(transcript)
+    if numbers is None:
+        return "Error: Text to number conversion returned None"
 
+    code_to_word_map = {v: k for k, v in word_to_code_map.items()}
+    return transcript_to_sentence(numbers, code_to_word_map)
+
+# Convert text to numerical representation
 def text2int(textnum, numwords={}):
-    units = ['Z600', 'O500','T000','T600','F600','F100','S220','S150','E300','N500',
-             'T500', 'E415', 'T410', 'T635', 'F635', 'F135', 'S235', 'S153', 'E235','N535']
+    units = ['Z600', 'O500', 'T000', 'T600', 'F600', 'F100', 'S220', 'S150', 'E300', 'N500',
+             'T500', 'E415', 'T410', 'T635', 'F635', 'F135', 'S235', 'S153', 'E235', 'N535']
     tens = ['', '', 'T537', 'T637', 'F637', 'F137', 'S230', 'S153', 'E230', 'N530']
     scales = ['H536', 'T253', 'M450', 'C600']
     ordinal_words = {'oh': 'Z600', 'first': 'O500', 'second': 'T000', 'third': 'T600', 'fourth': 'F600', 'fifth': 'F100',
-                     'sixth': 'S200','seventh': 'S150','eighth': 'E230', 'ninth': 'N500', 'twelfth': 'T410'}
+                     'sixth': 'S200', 'seventh': 'S150', 'eighth': 'E230', 'ninth': 'N500', 'twelfth': 'T410'}
     ordinal_endings = [('ieth', 'y'), ('th', '')]
     if not numwords:
         numwords['and'] = (1, 0)
@@ -304,18 +166,20 @@ def text2int(textnum, numwords={}):
             if scale > 100:
                 result += current
                 current = 0
-            onnumber = True
-            lastunit = False
+            lastunit = True
             lastscale = False
+            onnumber = True
         else:
             for ending, replacement in ordinal_endings:
                 if word.endswith(ending):
                     word = "%s%s" % (word[:-len(ending)], replacement)
 
-            if (not is_numword(word)) or (word == 'and' and not lastscale):
+            if not is_numword(word) or (word == 'and' and not lastscale):
                 if onnumber:
                     curstring += repr(result + current) + " "
-                curstring += word + " "
+                curstring += word
+                if word[-1] != '-':
+                    curstring += " "
                 result = current = 0
                 onnumber = False
                 lastunit = False
@@ -323,118 +187,102 @@ def text2int(textnum, numwords={}):
             else:
                 scale, increment = from_numword(word)
                 onnumber = True
-
-                if lastunit and (word not in scales):
+                if lastunit and (word in units or word in ordinal_words):
                     curstring += repr(result + current)
                     result = current = 0
-
                 if scale > 1:
                     current = max(1, current)
-
                 current = current * scale + increment
                 if scale > 100:
                     result += current
                     current = 0
-
-                lastscale = False
-                lastunit = False
-                if word in scales:
-                    lastscale = True
-                elif word in units:
-                    lastunit = True
+                lastunit = word in units
+                lastscale = word in scales
 
     if onnumber:
         curstring += repr(result + current)
 
     return curstring
 
+# Check if a word is a number
+def is_number(s):
+    try:
+        float(s.replace(',', ''))
+        return True
+    except ValueError:
+        return False
+
 # Convert sentence to transcript using Soundex
 def sentence_to_transcript(sentence, word_to_code_map):
-    words = sentence.split()
-    transcript_codes = []
-
-    for word in words:
-        if word not in word_to_code_map:
-            word_to_code_map[word] = soundex(word)
-        transcript_codes.append(word_to_code_map[word])
-    
-    transcript = ' '.join(transcript_codes)
-    return transcript
-
-# Convert transcript back to sentence using mapping
-def transcript_to_sentence(transcript, code_to_word_map):
-    codes = transcript.split()
-    sentence_words = []
+    with warnings.catch_warnings():
+        warnings.simplefilter("ignore")
+        from metaphone import doublemetaphone
 
-    for code in codes:
-        sentence_words.append(code_to_word_map.get(code, code))
-    
-    sentence = ' '.join(sentence_words)
-    return sentence
+        transcript = []
+        for word in sentence.split():
+            codes = doublemetaphone(word)
+            word_code = next((code for code in codes if code), None)
 
-# # Process the audio file
-# transcript = pipe("./odia_recorded/AUD-20240614-WA0004.wav")
-# text_value = transcript['text']
-# sentence = trn.transform(text_value)
-# replaced_words = replace_words(sentence)
-# processed_sentence = process_doubles(replaced_words)
-
-# input_sentence_1 = processed_sentence
-
-# Create empty mappings
-word_to_code_map = {}
-code_to_word_map = {}
-
-# Convert sentence to transcript
-# transcript_1 = sentence_to_transcript(input_sentence_1, word_to_code_map)
-
-# Convert transcript to numerical representation
-# numbers = text2int(transcript_1)
-
-# Create reverse mapping
-code_to_word_map = {v: k for k, v in word_to_code_map.items()}
-
-# Convert transcript back to sentence
-# reconstructed_sentence_1 = transcript_to_sentence(numbers, code_to_word_map)
-
-# demo=gr.Interface(
-#     fn=sel_lng, 
-      
-#     inputs=[
-        
-#         gr.Dropdown(["Hindi","Hindi-trans","Odiya","Odiya-trans"],value="Hindi",label="Select Language"),
-#         gr.Audio(source="microphone", type="filepath"),
-#         gr.Audio(source= "upload", type="filepath"),
-#         #gr.Audio(sources="upload", type="filepath"),
-#         #"state"
-#     ],
-#     outputs=[
-#         "textbox"
-# #        #"state"
-#     ],
-#     title="Automatic Speech Recognition",
-#     description = "Demo for Automatic Speech Recognition. Use microphone to record speech. Please press Record button. Initially it will take some time to load the model. The recognized text will appear in the output textbox",
-#       ).launch()
-
-######################################################    
-demo=gr.Interface(
-    fn=sel_lng, 
-      
-    inputs=[
-        
-        #gr.Dropdown(["Hindi","Hindi-trans","Odiya","Odiya-trans","Kannada","Kannada-trans","Telugu","Telugu-trans","Bangala","Bangala-trans"],value="Hindi",label="Select Language"),
-        gr.Dropdown(["Hindi","Hindi-trans","Odiya","Odiya-trans","Assamese-LM","Assamese-Model2"],value="Hindi",label="Select Language"),
-        gr.Audio(sources=["microphone","upload"], type="filepath"),
-        #gr.Audio(sources="upload", type="filepath"),
-        #"state"
-    ],
-    outputs=[
-        "textbox"
-#        #"state"
-    ],
-    allow_flagging="auto",
-    #flagging_options=["Language error", "English transliteration error", "Other"],
-    #flagging_callback=hf_writer,
-    title="Automatic Speech Recognition",
-    description = "Demo for Automatic Speech Recognition. Use microphone to record speech. Please press Record button. Initially it will take some time to load the model. The recognized text will appear in the output textbox",
-      ).launch()
+            if not word_code:
+                continue
+
+            if word_code not in word_to_code_map:
+                word_to_code_map[word] = word_code
+            transcript.append(word_code)
+
+        return ' '.join(transcript)
+
+# Convert transcript back to sentence using Soundex
+def transcript_to_sentence(transcript, code_to_word_map):
+    sentence = []
+    for code in transcript.split():
+        word = code_to_word_map.get(code, '')
+        if word:
+            sentence.append(word)
+    return ' '.join(sentence)
+
+# Set theme colors for Gradio interface
+theme_colors = {
+    "bg_color": "#0E1117",
+    "bg_secondary_color": "#161A25",
+    "input_color": "#161A25",
+    "input_text_color": "#C0C0BF",
+    "button_color": "#4A6AF2",
+    "button_primary_text_color": "#FFFFFF",
+    "button_secondary_color": "#A0A0A0",
+    "button_secondary_text_color": "#000000"
+}
+
+# Apply theme to Gradio blocks
+def apply_theme(demo):
+    demo.set_theme({
+        "background_color": theme_colors["bg_color"],
+        "secondary_background_color": theme_colors["bg_secondary_color"],
+        "input_background_color": theme_colors["input_color"],
+        "input_text_color": theme_colors["input_text_color"],
+        "button_primary_background_color": theme_colors["button_color"],
+        "button_primary_text_color": theme_colors["button_primary_text_color"],
+        "button_secondary_background_color": theme_colors["button_secondary_color"],
+        "button_secondary_text_color": theme_colors["button_secondary_text_color"]
+    })
+
+# Create Gradio interface
+with gr.Blocks() as demo:
+    apply_theme(demo)
+
+    gr.Markdown("<h1><center>Test</center></h1>")
+    with gr.Row():
+        language = gr.Dropdown(list(asr_models.keys()), label="Language", value="Hindi")
+        speech_input = gr.Audio(source="microphone", type="filepath", label="Speech")
+
+    text_output = gr.Textbox(label="Output")
+    submit_btn = gr.Button("Submit")
+
+    def process_audio(lang, speech):
+        transliterate = lang.endswith("-trans")
+        return transcribe(lang, speech, transliterate)
+
+    submit_btn.click(process_audio, inputs=[language, speech_input], outputs=text_output)
+
+# Launch the Gradio app on a different port
+demo.launch(server_port=7861)