Upload 14 files

src/inference.py

@@ -0,0 +1,233 @@
+import io
+import re
+import base64
+import numpy as np
+import traceback
+from typing import Union
+
+from TTS.utils.synthesizer import Synthesizer
+from aksharamukha.transliterate import process as aksharamukha_xlit
+from scipy.io.wavfile import write as scipy_wav_write
+
+import nltk
+import pysbd
+
+from .models.common import Language
+from .models.request import TTSRequest
+from .models.response import AudioFile, AudioConfig, TTSResponse, TTSFailureResponse
+from .utils.text import TextNormalizer
+from .utils.paragraph_handler import ParagraphHandler
+from src.postprocessor import PostProcessor
+
+class TextToSpeechEngine:
+    def __init__(
+        self,
+        models: dict,
+        allow_transliteration: bool = True,
+        enable_denoiser: bool = True,
+    ):
+        self.models = models
+        # TODO: Ability to instantiate models by accepting standard paths or auto-downloading
+        
+        code_mixed_found = False
+        if allow_transliteration:
+            # Initialize Indic-Xlit models for the languages corresponding to TTS models
+            from ai4bharat.transliteration import XlitEngine
+            xlit_langs = set()
+            
+            for lang in list(models):
+                if lang == 'en':
+                    continue # No need of any Indic-transliteration for English
+                
+                if '+' in lang:
+                    # If it's a code-mixed model like Hinglish, we need Hindi Xlit for non-English words
+                    # TODO: Make it mandatory irrespective of `allow_transliteration` boolean
+                    lang = lang.split('+')[1]
+                    code_mixed_found = True
+                xlit_langs.add(lang)
+            
+            self.xlit_engine = XlitEngine(xlit_langs, beam_width=6)
+        else:
+            self.xlit_engine = None
+
+        self.text_normalizer = TextNormalizer()
+        self.paragraph_handler = ParagraphHandler()
+        self.sent_seg = pysbd.Segmenter(language="en", clean=True)
+
+        self.orig_sr = 22050 # model.output_sample_rate
+        self.enable_denoiser = enable_denoiser
+        if enable_denoiser:
+            from src.postprocessor import Denoiser
+            self.target_sr = 16000
+            self.denoiser = Denoiser(self.orig_sr, self.target_sr)
+        else:
+            self.target_sr = self.orig_sr
+        
+        self.post_processor = PostProcessor(self.target_sr)
+
+        if code_mixed_found:
+            # Dictionary of English words
+            import enchant
+            from enchant.tokenize import get_tokenizer
+
+            self.enchant_dicts = {
+                "en_US": enchant.Dict("en_US"),
+                "en_GB": enchant.Dict("en_GB"),
+            }
+            self.enchant_tokenizer = get_tokenizer("en")
+
+    def concatenate_chunks(self, wav: np.ndarray, wav_chunk: np.ndarray):
+        # TODO: Move to utils
+        if type(wav_chunk) != np.ndarray:
+            wav_chunk = np.array(wav_chunk)
+        if wav is None:
+            return wav_chunk
+        return np.concatenate([wav, wav_chunk])
+
+    def infer_from_request(
+        self,
+        request: TTSRequest,
+        transliterate_roman_to_native: bool = True
+    ) -> TTSResponse:
+
+        config = request.config
+        lang = config.language.sourceLanguage
+        gender = config.gender
+
+        # If there's no separate English model, use the Hinglish one
+        if lang == "en" and lang not in self.models and "en+hi" in self.models:
+            lang = "en+hi"
+
+        if lang not in self.models:
+            return TTSFailureResponse(status_text="Unsupported language!")
+        
+        if lang == "brx" and gender == "male":
+            return TTSFailureResponse(status_text="Sorry, `male` speaker not supported for this language!")
+        
+        output_list = []
+
+        for sentence in request.input:
+            raw_audio = self.infer_from_text(sentence.source, lang, gender, transliterate_roman_to_native=transliterate_roman_to_native)
+            # Convert PCM to WAV
+            byte_io = io.BytesIO()
+            scipy_wav_write(byte_io, self.target_sr, raw_audio)
+            # Encode WAV fileobject as base64 for transmission via JSON
+            encoded_bytes = base64.b64encode(byte_io.read())
+            encoded_string = encoded_bytes.decode()
+            speech_response = AudioFile(audioContent=encoded_string)
+            
+            output_list.append(speech_response)
+
+        audio_config = AudioConfig(language=Language(sourceLanguage=lang))
+        return TTSResponse(audio=output_list, config=audio_config)
+    
+    def infer_from_text(
+        self,
+        input_text: str,
+        lang: str,
+        speaker_name: str,
+        transliterate_roman_to_native: bool = True
+    ) -> np.ndarray:
+        
+        # If there's no separate English model, use the Hinglish one
+        if lang == "en" and lang not in self.models and "en+hi" in self.models:
+            lang = "en+hi"
+        
+        input_text, primary_lang, secondary_lang = self.parse_langs_normalise_text(input_text, lang)
+
+        wav = None
+        paragraphs = self.paragraph_handler.split_text(input_text)
+
+        for paragraph in paragraphs:
+            paragraph = self.handle_transliteration(paragraph, primary_lang, transliterate_roman_to_native)
+            paras = []
+            for sent in self.sent_seg.segment(paragraph):
+                if sent.strip() and not re.match(r'^[_\W]+$', sent.strip()):
+                    paras.append(sent.strip())
+            paragraph = " ".join(paras)
+            
+            # Run Inference. TODO: Support for batch inference
+            wav_chunk = self.models[lang].tts(paragraph, speaker_name=speaker_name, style_wav="")
+
+            wav_chunk = self.postprocess_audio(wav_chunk, primary_lang, speaker_name)
+            # Concatenate current chunk with previous audio outputs
+            wav = self.concatenate_chunks(wav, wav_chunk)
+        return wav
+    
+    def parse_langs_normalise_text(self, input_text: str, lang: str) -> Union[str, str, str]:
+        # If there's no separate English model, use the Hinglish one if present
+        if lang == "en" and lang not in self.models and "en+hi" in self.models:
+            lang = "en+hi"
+
+        if lang == "en+hi": # Hinglish (English+Hindi code-mixed)
+            primary_lang, secondary_lang = lang.split('+')
+        else:
+            primary_lang = lang
+            secondary_lang = None
+
+        input_text = self.text_normalizer.normalize_text(input_text, primary_lang)
+        if secondary_lang:
+            # TODO: Write a proper `transliterate_native_words_using_eng_dictionary`
+            input_text = self.transliterate_native_words_using_spell_checker(input_text, secondary_lang)
+
+        return input_text, primary_lang, secondary_lang
+    
+    def handle_transliteration(self, input_text: str, primary_lang: str, transliterate_roman_to_native: bool) -> str:
+        if transliterate_roman_to_native and primary_lang != 'en':
+            input_text = self.transliterate_sentence(input_text, primary_lang)
+
+            # Manipuri was trained using the Central-govt's Bangla script
+            # So convert the words in native state-govt script to Eastern-Nagari
+            if primary_lang == "mni":
+                # TODO: Delete explicit-schwa
+                input_text = aksharamukha_xlit("MeeteiMayek", "Bengali", input_text)
+        return input_text
+        
+    def preprocess_text(
+        self,
+        input_text: str,
+        lang: str,
+        # speaker_name: str,
+        transliterate_roman_to_native: bool = True
+    ) -> np.ndarray:
+
+        input_text, primary_lang, secondary_lang = self.parse_langs_normalise_text(input_text, lang)
+        input_text = self.handle_transliteration(input_text, primary_lang, transliterate_roman_to_native)
+        return input_text
+
+    def postprocess_audio(self, wav_chunk, primary_lang, speaker_name):
+        if self.enable_denoiser:
+            wav_chunk = self.denoiser.denoise(wav_chunk)
+        wav_chunk = self.post_processor.process(wav_chunk, primary_lang, speaker_name)
+        return wav_chunk
+
+    def transliterate_native_words_using_spell_checker(self, input_text, lang):
+        tokens = [result[0] for result in self.enchant_tokenizer(input_text)]
+        pos_tags = [result[1] for result in nltk.tag.pos_tag(tokens)]
+
+        # Transliterate non-English Roman words to Indic
+        for word, pos_tag in zip(tokens, pos_tags):
+            if pos_tag == "NNP" or pos_tag == "NNPS":
+                # Enchant has many proper-nouns as well in its dictionary, don't know why.
+                # So if it's a proper-noun, always nativize
+                # FIXME: But NLTK's `averaged_perceptron_tagger` does not seem to be 100% accurate, it has false positives 🤦‍♂️ 
+                pass
+            elif self.enchant_dicts["en_US"].check(word) or self.enchant_dicts["en_GB"].check(word):
+                # TODO: Merge British and American dicts into 1 somehow
+                continue
+            
+            # Convert "Ram's" -> "Ram". TODO: Think what are the failure cases
+            word = word.split("'")[0]
+
+            transliterated_word = self.transliterate_sentence(word, lang)
+            input_text = input_text.replace(word, transliterated_word, 1)
+        return input_text
+
+    def transliterate_sentence(self, input_text, lang):
+        if not self.xlit_engine:
+            return input_text
+
+        if lang == "raj":
+            lang = "hi" # Approximate
+        
+        return self.xlit_engine.translit_sentence(input_text, lang)

src/models/common.py

@@ -0,0 +1,11 @@
+from pydantic import BaseModel, validator
+
+
+class Language(BaseModel):
+    sourceLanguage: str
+
+    # @validator('sourceLanguage', pre=True)
+    # def blank_string_in_language(cls, value, field):
+    #     if value == "":
+    #         raise ValueError('sourceLanguage cannot be empty')
+    #     return value

src/models/request.py

@@ -0,0 +1,41 @@
+from typing import List
+
+from pydantic import BaseModel, validator
+
+from .common import Language
+
+SUPPORTED_GENDERS = {'male', 'female'}
+
+
+class Sentence(BaseModel):
+    source: str
+
+    # @validator('source', pre=True)
+    # def blank_string_in_source(cls, value, field):
+    #     if value == "":
+    #         raise ValueError('source cannot be empty')
+    #     return value
+
+
+class TTSConfig(BaseModel):
+    language: Language
+    gender: str
+
+    # @validator('gender', pre=True)
+    # def blank_string_in_gender(cls, value, field):
+    #     if value == "":
+    #         raise ValueError('gender cannot be empty')
+    #     if value not in SUPPORTED_GENDERS:
+    #         raise ValueError('Unsupported gender value')
+    #     return value
+
+
+class TTSRequest(BaseModel):
+    input: List[Sentence]
+    config: TTSConfig
+
+    # @validator('input', pre=True)
+    # def input_cannot_be_empty(cls, value, field):
+    #     if len(value) < 1:
+    #         raise ValueError('input cannot be empty')
+    #     return value

src/models/response.py

@@ -0,0 +1,26 @@
+from typing import List
+
+from pydantic import BaseModel
+
+from .common import Language
+
+
+class AudioFile(BaseModel):
+    audioContent: str
+
+
+class AudioConfig(BaseModel):
+    language: Language
+    audioFormat: str = 'wav'
+    encoding: str = 'base64'
+    samplingRate: int = 22050
+
+
+class TTSResponse(BaseModel):
+    audio: List[AudioFile]
+    config: AudioConfig
+
+
+class TTSFailureResponse(BaseModel):
+    status: str = 'ERROR'
+    status_text: str

src/postprocessor/__init__.py

@@ -0,0 +1,3 @@
+from .postprocessor import PostProcessor
+from .denoiser import Denoiser
+from .vad import VoiceActivityDetection

src/postprocessor/denoiser.py

@@ -0,0 +1,24 @@
+import torch
+import librosa
+import numpy as np
+
+class Denoiser:
+
+    def __init__(self, orig_sr:int, target_sr:int):
+        self.orig_sr = orig_sr
+        self.target_sr = target_sr
+        self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
+
+        from asteroid.models import BaseModel as AsteroidBaseModel
+        self.model = AsteroidBaseModel.from_pretrained("JorisCos/DCCRNet_Libri1Mix_enhsingle_16k").to(self.device)
+    
+    def denoise(self, wav):
+        if type(wav) != np.ndarray:
+            wav = np.array(wav)
+        
+        if len(wav.shape) > 1:
+            wav = np.mean(wav, axis=1)
+        wav = librosa.resample(wav, orig_sr=self.orig_sr, target_sr=self.target_sr)
+        wav = torch.Tensor(wav.reshape(1, 1, wav.shape[0])).float().to(self.device)
+        wav = self.model.separate(wav)[0][0] #(batch, channels, time) -> (time)
+        return wav.cpu().detach().numpy()

src/postprocessor/postprocessor.py

@@ -0,0 +1,46 @@
+import os
+import ffmpeg
+import librosa
+import numpy as np
+import soundfile as sf
+import tempfile
+
+from .vad import VoiceActivityDetection
+
+
+class PostProcessor:
+
+    def __init__(self, target_sr:int):
+        self.target_sr = target_sr
+        self.vad = VoiceActivityDetection()
+
+    def set_tempo(self, wav:np.ndarray, atempo:str ='1'):
+        with tempfile.TemporaryDirectory() as tmpdirname:
+            inpath = os.path.join(tmpdirname, 'input.wav')
+            outpath = inpath.replace('input.wav', 'output.wav')
+            sf.write(inpath, wav, self.target_sr)
+            in_stream = ffmpeg.input(inpath)
+            audio_stream = ffmpeg.filter_(in_stream, 'atempo', atempo)
+            audio_stream = audio_stream.output(outpath)
+            ffmpeg.run(audio_stream, overwrite_output=True)
+            wav, _ = librosa.load(outpath, sr=self.target_sr)
+        return wav
+    
+    def trim_silence(self, wav:np.ndarray):
+        return self.vad.process(wav, sc_threshold=40)
+
+    def process(self, wav, lang:str, gender:str):
+        if type(wav) != np.ndarray:
+            wav = np.array(wav)
+
+        if (lang == "te") and (gender=='female'):  # Telugu female speaker slow down
+            wav = self.set_tempo(wav, '0.85')
+            wav = self.trim_silence(wav) 
+        elif (lang == 'mr') and (gender=='female'):  # Marathi female speaker speed up
+            wav = self.trim_silence(wav)
+            wav = self.set_tempo(wav, '1.15')
+        elif (lang == 'gu'):  # Gujarati speaker speed up
+            # wav = trim_silence(wav)
+            wav = self.set_tempo(wav, '1.20')
+
+        return wav

src/postprocessor/vad.py

@@ -0,0 +1,87 @@
+#! /usr/bin/env python
+# encoding: utf-8
+'''
+MIT License
+
+Copyright (c) 2018 Mauricio
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
+
+Adapted from https://github.com/mauriciovander/silence-removal/blob/master/vad.py
+'''
+import numpy
+
+class VoiceActivityDetection:
+
+    def __init__(self):
+        self.__step = 160
+        self.__buffer_size = 160 
+        self.__buffer = numpy.array([],dtype=numpy.int16)
+        self.__out_buffer = numpy.array([],dtype=numpy.int16)
+        self.__n = 0
+        self.__VADthd = 0.
+        self.__VADn = 0.
+        self.__silence_counter = 0
+
+    # Voice Activity Detection
+    # Adaptive threshold
+    def vad(self, _frame, sc_threshold=20):
+        frame = numpy.array(_frame) ** 2.
+        result = True
+        threshold = 0.2
+        thd = numpy.min(frame) + numpy.ptp(frame) * threshold
+        self.__VADthd = (self.__VADn * self.__VADthd + thd) / float(self.__VADn + 1.)
+        self.__VADn += 1.
+
+        if numpy.mean(frame) <= self.__VADthd:
+            self.__silence_counter += 1
+        else:
+            self.__silence_counter = 0
+        if self.__silence_counter > sc_threshold:
+            result = False
+        return result
+
+    # Push new audio samples into the buffer.
+    def add_samples(self, data):
+        self.__buffer = numpy.append(self.__buffer, data)
+        result = len(self.__buffer) >= self.__buffer_size
+        # print('__buffer size %i'%self.__buffer.size)
+        return result
+
+    # Pull a portion of the buffer to process
+    # (pulled samples are deleted after being
+    # processed
+    def get_frame(self):
+        window = self.__buffer[:self.__buffer_size]
+        self.__buffer = self.__buffer[self.__step:]
+        # print('__buffer size %i'%self.__buffer.size)
+        return window
+
+    # Adds new audio samples to the internal
+    # buffer and process them
+    def process(self, data, sc_threshold):
+        self.__buffer = numpy.array([],dtype=numpy.int16)
+        self.__out_buffer = numpy.array([],dtype=numpy.int16)
+        if self.add_samples(data):
+            while len(self.__buffer) >= self.__buffer_size:
+                # Framing
+                window = self.get_frame()
+                if self.vad(window, sc_threshold):  # speech frame
+                	self.__out_buffer = numpy.append(self.__out_buffer, window)
+        return self.__out_buffer

src/utils/alphabet2phone.json

@@ -0,0 +1 @@
+{"a": "aey", "b": "bee", "c": "see", "d": "dee", "e": "eee", "f": "eff", "g": "jee", "h": "ech", "i": "aai", "j": "jay", "k": "kay", "l": "ell", "m": "em", "n": "en", "o": "oh", "p": "pee", "q": "kyuu", "r": "aar", "s": "es", "t": "tea", "u": "you", "v": "vee", "w": "doubleu", "x": "ex", "y": "why", "z": "zedd"}

src/utils/paragraph_handler.py

@@ -0,0 +1,43 @@
+#! /usr/bin/env python
+# encoding: utf-8
+
+import re
+
+non_chars_regex = re.compile(r'[^\w]')
+
+class ParagraphHandler():
+
+    def __init__(self, max_text_len=512):
+        self.L = max_text_len
+    
+    def split_text(self, text:str, delimiter='.'):
+        '''Splits text at delimiter into paragraphs of max. length self.L'''
+        delimiter = ' ' if delimiter not in text else delimiter
+        if delimiter not in text:
+            return [text]
+        
+        paragraphs = []
+        l_pos, r_pos = 0, 0
+        while r_pos < len(text):
+            r_pos = l_pos + self.L
+            if r_pos >= len(text): # append last paragraph. 
+                paragraphs.append(text[l_pos:len(text)])
+                break
+            while delimiter is not None and text[r_pos] != delimiter and r_pos > l_pos and r_pos > 0: # find nearest delimiter < r_pos to split paragraph at.
+                r_pos -= 1
+            extracted_paragraph = text[l_pos:r_pos+1]
+            extracted_paragraph_without_special_chars = non_chars_regex.sub('', extracted_paragraph)
+            if extracted_paragraph_without_special_chars:
+                paragraphs.append(extracted_paragraph)
+            l_pos = r_pos + 1  # handle next paragraph
+        return paragraphs
+
+
+if __name__ == '__main__':
+    text = "The following are quotes from A.P.J. Abdul Kalam. To succeed in your mission, you must have single-minded devotion to your goal. Look at the sky. We are not alone. The whole universe is friendly to us and conspires only to give the best to those who dream and work. The youth need to be enabled to become job generators from job seekers. If four things are followed - having a great aim, acquiring knowledge, hard work, and perseverance - then anything can be achieved. Where there is righteousness in the heart, there is beauty in the character. When there is beauty in the character, there is harmony in the home. When there is harmony in the home, there is order in the nation. When there is order in the nation, there is peace in the world. Great teachers emanate out of knowledge, passion and compassion. Let me define a leader. He must have vision and passion and not be afraid of any problem. Instead, he should know how to defeat it. Most importantly, he must work with integrity."
+    print('LENGTH: ', len(text))  # 988
+
+    paragraph_handler = ParagraphHandler()
+    paragraphs = paragraph_handler.split_text(text)
+    for p in paragraphs:
+        print(len(p), p)

src/utils/symbols.json

@@ -0,0 +1,155 @@
+{
+    "₹": {
+        "as": "Rupees",
+        "bn": "রুপি",
+        "brx": "Rupees",
+        "en": "Rupees",
+        "gu": "રૂપિયા",
+        "hi": "रुपये",
+        "kn": "ರೂಪಾಯಿ",
+        "ml": "രൂപ",
+        "mni": "Rupees",
+        "mr": "रुपये",
+        "or": "ଟଙ୍କ।",
+        "pa": "ਰੁਪਏ",
+        "raj": "रीप्या",
+        "ta": "ரூபாய்",
+        "te": "రూపాయలు"
+    },
+    "@": {
+        "as": "আঁ ট",
+        "bn": "আট",
+        "brx": "at",
+        "en": "at",
+        "gu": "આત્‌",
+        "hi": "आट",
+        "kn": "ಅಟ್‌",
+        "ml": "ആറ്റ്‌",
+        "mni": "ꯑꯊ",
+        "mr": "आट",
+        "or": "ଆଟ",
+        "pa": "ਆਤ",
+        "raj": "आट",
+        "ta": "ஆட்‌",
+        "te": "ఆ ట"
+    },
+    ".": {
+        "as": "ডোট",
+        "bn": "ডোট",
+        "brx": "dot",
+        "en": "dot",
+        "gu": "ડોટ",
+        "hi": "डोट",
+        "kn": "dot",
+        "ml": "ഡോട്ട്‌",
+        "mni": "ꯗꯣꯇ",
+        "mr": "डॉट",
+        "or": "ଡୋଟ୍‌",
+        "pa": "ਡੋਟ",
+        "raj": "डॉट",
+        "ta": "டாட்‌",
+        "te": "డాట్‌"
+    },
+    "/": {
+        "as": "শ্লাচ",
+        "bn": "স্লাশ",
+        "brx": "स्लाश",
+        "en": "slash",
+        "gu": "સ્લેશ",
+        "hi": "सलाश",
+        "kn": "ಸ್ಲಾಶ್‌",
+        "ml": "സ്ലാഷ്‌",
+        "mni": "ꯁ꯭ꯂꯦꯁ",
+        "mr": "सलाश",
+        "or": "ସ୍ଲାଶ୍‌",
+        "pa": "slash", 
+        "raj": "स्लाश",
+        "ta": "ஸ்லாஷ்‌",
+        "te": "స్లాష్‌"
+    },
+    ":": {
+        "as": "কোলন",
+        "bn": "কোলন",
+        "brx": "कोलन",
+        "en": "colon",
+        "gu": "કોલન",
+        "hi": "कोलन",
+        "kn": "ಕೋಲನ್‌",
+        "ml": "കോളൻ",
+        "mni": "ꯀꯣꯂꯦꯟ",
+        "mr": "कोलन",
+        "or": "କୋଲୋନ",
+        "pa": "ਕੌਲੋਨ",
+        "raj": "कोलन",
+        "ta": "கோலன்‌",
+        "te": "కోలన్‌"
+    },
+    "+": {
+        "as": "প্লাচ",
+        "bn": "প্লাস",
+        "brx": "प्लस",
+        "en": "plus",
+        "gu": "પ્લસ",
+        "hi": "प्लस",
+        "kn": "ಪ್ಲಸ್‌",
+        "ml": "പ്ലസ്‌",
+        "mni": "ꯄ꯭ꯂꯁ",
+        "mr": "प्लॅस",
+        "or": "ପ୍ଲସ୍‌",
+        "pa": "ਪਲੱਸ",
+        "raj": "प्लस",
+        "ta": "பிளஸ்‌",
+        "te": "ప్లస్‌"
+    },
+    "-": {
+        "as": "ডাছ",
+        "bn": "ডাছ",
+        "brx": "दाश",
+        "en": "dash",
+        "gu": "ડાશ",
+        "hi": "दाश",
+        "kn": "ದಾಶು",
+        "ml": "ദാശ",
+        "mni": "ꯗꯥꯁ",
+        "mr": "दाश",
+        "or": "ଦାଶ",
+        "pa": "ਮਾਈਨਉਸ",
+        "raj": "दाश",
+        "ta": "டாஷ்‌",
+        "te": "డాష్‌"
+    },
+    "www": {
+        "as": "ডাব্লিঅডাব্লিওডাব্লিও",
+        "bn": "ডাব্লিউডাব্লিউডাব্লিউ",
+        "brx": "डबलयु डबलयु डबलयु",
+        "en": "doubleyou doubleyou doubleyou",
+        "gu": "ડબલ્યું ડબલ્યું ડબલ્યું",
+        "hi": "डबल्यू डबल्यू डबल्यू",
+        "kn": "ಡುಬ್ಲ್ಯುಡುಬ್ಲ್ಯುಡುಬ್ಲ್ಯು ",
+        "ml": "ഡബ്ലിയൂ ഡബ്ലിയൂ ഡബ്ലിയൂ",
+        "mni": "ꯗꯕꯜꯌꯨꯗꯕꯜꯌꯨꯗꯕꯜꯌꯨ",
+        "mr": "डब्ल्यू डब्ल्यू डब्ल्यू",
+        "or": "ଡବ୍ଲିୟୁଡବ୍ଲିୟୁଡବ୍ଲିୟୁ",
+        "pa": "ਡਬਲਿਊ ਡਬਲਿਊ ਡਬਲਿਊ",
+        "raj": "डब्ल्यू डब्ल्यू डब्ल्यू",
+        "ta": "டபிளியூ டபிளியூ டபிளியூ",
+        "te": "డబుల్యూడబుల్యూడబుల్యూ"
+    },
+    "%": {
+        "as": "শতাংশ",
+        "bn": "শতাংশ",
+        "brx": "percent",
+        "en": "percent",
+        "gu": "ટકા",
+        "hi": "प्रतिशत",
+        "kn": "ಶೇಕಡಾ",
+        "ml": "ശതമാനം",
+        "mni": "ꯆꯥꯗꯥ ꯆꯥꯗꯥ ꯴",
+        "mr": "टक्के",
+        "or": "ଶତକଡା",
+        "pa": "ਪ੍ਰਤੀਸ਼ਤ",
+        "raj": "percent",
+        "ta": "சதவீதம்",
+        "te": "శాతం"
+    }
+}

src/utils/text.py

@@ -0,0 +1,212 @@
+import os
+PWD = os.path.dirname(__file__)
+import re
+import regex
+import json
+import traceback
+
+from nemo_text_processing.text_normalization.normalize import Normalizer
+from indic_numtowords import num2words, supported_langs
+from .translator import GoogleTranslator
+
+indic_acronym_matcher = regex.compile(r"([\p{L}\p{M}]+\.\s*){2,}")
+
+# short_form_regex = re.compile(r'\b[A-Z\.]{2,}s?\b')
+# def get_shortforms_from_string(text):
+#     return short_form_regex.findall(text)
+
+short_form_regex = re.compile(r"\b([A-Z][\.\s]+)+([A-Z])?\b")
+eng_consonants_regex = re.compile(r"\b[BCDFGHJKLMNPQRSTVWXZbcdfghjklmnpqrstvwxz]+\b")
+def get_shortforms_from_string(text):
+  dotted_shortforms = [m.group() for m in re.finditer(short_form_regex, text)]
+  non_dotted_shortforms = [m.group() for m in re.finditer(eng_consonants_regex, text)]
+  return dotted_shortforms + non_dotted_shortforms
+
+decimal_str_regex = re.compile("\d{1,3}(?:(?:,\d{2,3}){1,3}|(?:\d{1,7}))?(?:\.\d+)")
+def get_all_decimals_from_string(text):
+  return decimal_str_regex.findall(text)
+
+num_str_regex = re.compile("\d{1,3}(?:(?:,\d{2,3}){1,3}|(?:\d{1,7}))?(?:\.\d+)?")
+def get_all_numbers_from_string(text):
+  return num_str_regex.findall(text)
+
+multiple_stops_regex = r'\.\.+'
+def replace_multiple_stops(text):
+  return re.sub(multiple_stops_regex, '.', text) 
+
+date_generic_match_regex = re.compile("(?:[^0-9]\d*[./-]\d*[./-]\d*)")
+date_str_regex = re.compile("(?:\d{1,2}[./-]\d{1,2}[./-]\d{2,4})|(?:\d{2,4}[./-]\d{1,2}[./-]\d{1,2})")  # match like dd/mm/yyyy or dd-mm-yy or yyyy.mm.dd or yy/mm/dd
+def get_all_dates_from_string(text):
+  candidates = date_generic_match_regex.findall(text)
+  candidates = [c.replace(' ', '') for c in candidates]
+  candidates = [c for c in candidates if len(c) <= 10]  # Prune invalid dates
+  candidates = ' '.join(candidates)
+  return date_str_regex.findall(candidates)
+
+def get_decimal_substitution(decimal):
+  decimal_parts = decimal.split('.')
+  l_part = decimal_parts[0]
+  r_part = ""
+  for part in decimal_parts[1:]:
+    r_part += ' '.join(list(part))  # space between every digit after decimal point
+  decimal_sub = l_part + " point " + r_part 
+  decimal_sub = decimal_sub.strip()
+  return decimal_sub
+
+email_regex = r'[\w.+-]+@[\w-]+\.[\w.-]+'
+url_regex = r'((?:\w+://)?\w+\.\w+\.\w+/?[\w\.\?=#]*)|(\w*.com/?[\w\.\?=#]*)'
+currency_regex = r"\₹\ ?[+-]?[0-9]{1,3}(?:,?[0-9])*(?:\.[0-9]{1,2})?"
+phone_regex = r'\+?\d[ \d-]{6,12}\d'
+
+
+
+class TextNormalizer:
+  def __init__(self):
+    self.translator = GoogleTranslator()
+    self.normalizer = Normalizer(input_case='cased', lang='en')
+    self.symbols2lang2word = json.load(open(os.path.join(PWD, "symbols.json"), encoding="utf-8"))
+    self.alphabet2phone = json.load(open(os.path.join(PWD, "alphabet2phone.json"), encoding="utf-8"))
+  
+  def normalize_text(self, text, lang):
+    text = text.replace("।", ".").replace("|", ".").replace("꯫", ".").strip()
+    text = self.expand_shortforms(text, lang)
+    text = self.normalize_decimals(text, lang)
+    text = self.replace_punctutations(text, lang)
+    text = self.convert_dates_to_words(text, lang)
+    text = self.convert_symbols_to_words(text, lang)
+    text = self.convert_numbers_to_words(text, lang)
+    return text
+  
+  def normalize_decimals(self, text, lang):
+    decimal_strs = get_all_decimals_from_string(text)
+    if not decimal_strs:
+      return text
+    decimals = [str(decimal_str.replace(',', '')) for decimal_str in decimal_strs]
+    decimal_substitutions = [get_decimal_substitution(decimal) for decimal in decimals]
+    for decimal_str, decimal_sub in zip(decimal_strs, decimal_substitutions):
+      text = text.replace(decimal_str, decimal_sub)
+    return text   
+
+  def replace_punctutations(self, text, lang):
+    text = replace_multiple_stops(text)
+    if lang not in ['brx', 'or']:
+      text = text.replace('।', '.')
+      if text[-1] not in ['.', '!', '?', ',', ':', ';']:
+        text = text + ' .'
+    else:
+      text = text.replace('.', '।')
+    text = text.replace('|', '.')
+    for bracket in ['(', ')', '{', '}', '[', ']']:
+      text = text.replace(bracket, ',')
+    # text = text.replace(':', ',').replace(';',',')
+    text = text.replace(';',',')
+    return text
+  
+  def convert_numbers_to_words(self, text, lang):
+    num_strs = get_all_numbers_from_string(text)
+    if not num_strs:
+      return text
+    
+    # TODO: If it is a large integer without commas (say >5 digits), spell it out numeral by numeral
+    # NOTE: partially handled by phones
+    numbers = [int(num_str.replace(',', '')) for num_str in num_strs]
+    
+    if lang in supported_langs:
+      # print(lang, numbers)
+      num_words = [num2words(num, lang=lang) for num in numbers]
+    else: # Fallback, converting to Indian-English, followed by NMT
+      try:
+        num_words = [num2words(num, lang="en") for num in numbers]
+        translated_num_words = [self.translator(text=num_word, from_lang="en", to_lang=lang) for num_word in num_words]
+        # TODO: Cache the results?
+        num_words = translated_num_words
+      except:
+        traceback.print_exc()
+    
+    for num_str, num_word in zip(num_strs, num_words):
+      text = text.replace(num_str, ' '+num_word+' ', 1)
+    return text.replace("  ", ' ')
+
+  def convert_dates_to_words(self, text, lang):
+    date_strs = get_all_dates_from_string(text)
+    if not date_strs:
+      return text
+    for date_str in date_strs:
+      normalized_str = self.normalizer.normalize(date_str, verbose=False, punct_post_process=True)
+      if lang in ['brx', 'en']:  # no translate
+        translated_str = normalized_str
+      else:
+        translated_str = self.translator(text=normalized_str, from_lang="en", to_lang=lang)
+      text = text.replace(date_str, translated_str)
+    return text
+
+  def expand_phones(self, item):
+    return ' '.join(list(item))
+  
+  def find_valid(self, regex_str, text):
+    items = re.findall(regex_str, text)
+    return_items = []
+    for item in items:
+      if isinstance(item, tuple):
+        for subitem in item:
+          if len(subitem) > 0:
+            return_items.append(subitem)
+            break  # choose first valid sub item
+      elif len(item) > 0:
+        return_items.append(item)
+    return return_items
+  
+  def convert_symbols_to_words(self, text, lang):
+    symbols = self.symbols2lang2word.keys()
+    emails = self.find_valid(email_regex, text)
+    # urls = re.findall(r'(?:\w+://)?\w+\.\w+\.\w+/?[\w\.\?=#]*', text)
+    urls = self.find_valid(url_regex, text)
+    # print('URLS', urls)
+    for item in emails + urls:
+      item_norm = item
+      for symbol in symbols:
+        item_norm = item_norm.replace(symbol, f' {self.symbols2lang2word[symbol][lang]} ')
+      text = text.replace(item, item_norm)
+    
+    currencies = self.find_valid(currency_regex, text)
+    for item in currencies:
+      item_norm = item.replace('₹','') + '₹'  # Pronounce after numerals
+      for symbol in symbols:
+        item_norm = item_norm.replace(symbol, f' {self.symbols2lang2word[symbol][lang]} ')
+      text = text.replace(item, item_norm)
+    
+    phones = self.find_valid(phone_regex, text)
+    for item in phones:
+      item_norm = item.replace('-', ' ')
+      for symbol in symbols:
+        item_norm = item_norm.replace(symbol, f' {self.symbols2lang2word[symbol][lang]} ')
+      item_norm = self.expand_phones(item_norm)
+      text = text.replace(item, item_norm)
+    
+    # percentage
+    text = text.replace('%', self.symbols2lang2word['%'][lang])
+    
+    return text
+
+  def convert_char2phone(self, char):
+        return self.alphabet2phone[char.lower()] if char.lower() in self.alphabet2phone else ''
+  
+  def expand_shortforms(self, text, lang):
+    if lang!='en':
+      # Remove dots, as it speaks out like each letter is separate sentence
+      # Example: अई. अई. टी. -> अई अई टी
+      for match in regex.finditer(indic_acronym_matcher, text):
+        match = match.group()
+        match_without_dot = match.replace('.', ' ')
+        text = text.replace(match, match_without_dot)
+      return text
+    
+    shortforms = get_shortforms_from_string(text)
+    for shortform in shortforms:
+        shortform = shortform.strip()
+        if shortform == 'I' or shortform == "A":
+          # Skip if valid English words
+          continue
+        expanded = ' '.join([self.convert_char2phone(char) for char in shortform])
+        text = text.replace(shortform, expanded, 1)
+    return text

src/utils/translator.py

@@ -0,0 +1,27 @@
+class GoogleTranslator:
+  def __init__(self):
+    from translators.server import google, _google
+    self._translate = google
+
+    google("Testing...")
+    self.supported_languages = set(_google.language_map['en'])
+    self.custom_lang_map = {
+        "mni": "mni-Mtei",
+        "raj": "hi",
+    }
+
+  def translate(self, text, from_lang, to_lang):
+    if from_lang in self.custom_lang_map:
+      from_lang = self.custom_lang_map[from_lang]
+    elif from_lang not in self.supported_languages:
+      return text
+    
+    if to_lang in self.custom_lang_map:
+      to_lang = self.custom_lang_map[to_lang]
+    elif to_lang not in self.supported_languages:
+      return text
+    
+    return self._translate(text, from_language=from_lang, to_language=to_lang)
+  
+  def __call__(self, **kwargs):
+    return self.translate(**kwargs)