import torch import torch.nn as nn import numpy as np import pandas as pd import random import sys import os import json import enum import traceback import re #F_DIR = os.path.dirname(os.path.realpath(__file__)) F_DIR = '/home/user/app/ttsv/checkpoints/' class XlitError(enum.Enum): lang_err = "Unsupported langauge ID requested ;( Please check available languages." string_err = "String passed is incompatable ;(" internal_err = "Internal crash ;(" unknown_err = "Unknown Failure" loading_err = "Loading failed ;( Check if metadata/paths are correctly configured." ##=================== Network ================================================== class Encoder(nn.Module): def __init__( self, input_dim, embed_dim, hidden_dim, rnn_type="gru", layers=1, bidirectional=False, dropout=0, device="cpu", ): super(Encoder, self).__init__() self.input_dim = input_dim # src_vocab_sz self.enc_embed_dim = embed_dim self.enc_hidden_dim = hidden_dim self.enc_rnn_type = rnn_type self.enc_layers = layers self.enc_directions = 2 if bidirectional else 1 self.device = device self.embedding = nn.Embedding(self.input_dim, self.enc_embed_dim) if self.enc_rnn_type == "gru": self.enc_rnn = nn.GRU( input_size=self.enc_embed_dim, hidden_size=self.enc_hidden_dim, num_layers=self.enc_layers, bidirectional=bidirectional, ) elif self.enc_rnn_type == "lstm": self.enc_rnn = nn.LSTM( input_size=self.enc_embed_dim, hidden_size=self.enc_hidden_dim, num_layers=self.enc_layers, bidirectional=bidirectional, ) else: raise Exception("XlitError: unknown RNN type mentioned") def forward(self, x, x_sz, hidden=None): """ x_sz: (batch_size, 1) - Unpadded sequence lengths used for pack_pad """ batch_sz = x.shape[0] # x: batch_size, max_length, enc_embed_dim x = self.embedding(x) ## pack the padded data # x: max_length, batch_size, enc_embed_dim -> for pack_pad x = x.permute(1, 0, 2) x = nn.utils.rnn.pack_padded_sequence(x, x_sz, enforce_sorted=False) # unpad # output: packed_size, batch_size, enc_embed_dim # hidden: n_layer**num_directions, batch_size, hidden_dim | if LSTM (h_n, c_n) output, hidden = self.enc_rnn( x ) # gru returns hidden state of all timesteps as well as hidden state at last timestep ## pad the sequence to the max length in the batch # output: max_length, batch_size, enc_emb_dim*directions) output, _ = nn.utils.rnn.pad_packed_sequence(output) # output: batch_size, max_length, hidden_dim output = output.permute(1, 0, 2) return output, hidden def get_word_embedding(self, x): """ """ x_sz = torch.tensor([len(x)]) x_ = torch.tensor(x).unsqueeze(0).to(dtype=torch.long) # x: 1, max_length, enc_embed_dim x = self.embedding(x_) ## pack the padded data # x: max_length, 1, enc_embed_dim -> for pack_pad x = x.permute(1, 0, 2) x = nn.utils.rnn.pack_padded_sequence(x, x_sz, enforce_sorted=False) # unpad # output: packed_size, 1, enc_embed_dim # hidden: n_layer**num_directions, 1, hidden_dim | if LSTM (h_n, c_n) output, hidden = self.enc_rnn( x ) # gru returns hidden state of all timesteps as well as hidden state at last timestep out_embed = hidden[0].squeeze() return out_embed class Decoder(nn.Module): def __init__( self, output_dim, embed_dim, hidden_dim, rnn_type="gru", layers=1, use_attention=True, enc_outstate_dim=None, # enc_directions * enc_hidden_dim dropout=0, device="cpu", ): super(Decoder, self).__init__() self.output_dim = output_dim # tgt_vocab_sz self.dec_hidden_dim = hidden_dim self.dec_embed_dim = embed_dim self.dec_rnn_type = rnn_type self.dec_layers = layers self.use_attention = use_attention self.device = device if self.use_attention: self.enc_outstate_dim = enc_outstate_dim if enc_outstate_dim else hidden_dim else: self.enc_outstate_dim = 0 self.embedding = nn.Embedding(self.output_dim, self.dec_embed_dim) if self.dec_rnn_type == "gru": self.dec_rnn = nn.GRU( input_size=self.dec_embed_dim + self.enc_outstate_dim, # to concat attention_output hidden_size=self.dec_hidden_dim, # previous Hidden num_layers=self.dec_layers, batch_first=True, ) elif self.dec_rnn_type == "lstm": self.dec_rnn = nn.LSTM( input_size=self.dec_embed_dim + self.enc_outstate_dim, # to concat attention_output hidden_size=self.dec_hidden_dim, # previous Hidden num_layers=self.dec_layers, batch_first=True, ) else: raise Exception("XlitError: unknown RNN type mentioned") self.fc = nn.Sequential( nn.Linear(self.dec_hidden_dim, self.dec_embed_dim), nn.LeakyReLU(), # nn.Linear(self.dec_embed_dim, self.dec_embed_dim), nn.LeakyReLU(), # removing to reduce size nn.Linear(self.dec_embed_dim, self.output_dim), ) ##----- Attention ---------- if self.use_attention: self.W1 = nn.Linear(self.enc_outstate_dim, self.dec_hidden_dim) self.W2 = nn.Linear(self.dec_hidden_dim, self.dec_hidden_dim) self.V = nn.Linear(self.dec_hidden_dim, 1) def attention(self, x, hidden, enc_output): """ x: (batch_size, 1, dec_embed_dim) -> after Embedding enc_output: batch_size, max_length, enc_hidden_dim *num_directions hidden: n_layers, batch_size, hidden_size | if LSTM (h_n, c_n) """ ## perform addition to calculate the score # hidden_with_time_axis: batch_size, 1, hidden_dim ## hidden_with_time_axis = hidden.permute(1, 0, 2) ## replaced with below 2lines hidden_with_time_axis = ( torch.sum(hidden, axis=0) if self.dec_rnn_type != "lstm" else torch.sum(hidden[0], axis=0) ) # h_n hidden_with_time_axis = hidden_with_time_axis.unsqueeze(1) # score: batch_size, max_length, hidden_dim score = torch.tanh(self.W1(enc_output) + self.W2(hidden_with_time_axis)) # attention_weights: batch_size, max_length, 1 # we get 1 at the last axis because we are applying score to self.V attention_weights = torch.softmax(self.V(score), dim=1) # context_vector shape after sum == (batch_size, hidden_dim) context_vector = attention_weights * enc_output context_vector = torch.sum(context_vector, dim=1) # context_vector: batch_size, 1, hidden_dim context_vector = context_vector.unsqueeze(1) # attend_out (batch_size, 1, dec_embed_dim + hidden_size) attend_out = torch.cat((context_vector, x), -1) return attend_out, attention_weights def forward(self, x, hidden, enc_output): """ x: (batch_size, 1) enc_output: batch_size, max_length, dec_embed_dim hidden: n_layer, batch_size, hidden_size | lstm: (h_n, c_n) """ if (hidden is None) and (self.use_attention is False): raise Exception( "XlitError: No use of a decoder with No attention and No Hidden" ) batch_sz = x.shape[0] if hidden is None: # hidden: n_layers, batch_size, hidden_dim hid_for_att = torch.zeros( (self.dec_layers, batch_sz, self.dec_hidden_dim) ).to(self.device) elif self.dec_rnn_type == "lstm": hid_for_att = hidden[1] # c_n # x (batch_size, 1, dec_embed_dim) -> after embedding x = self.embedding(x) if self.use_attention: # x (batch_size, 1, dec_embed_dim + hidden_size) -> after attention # aw: (batch_size, max_length, 1) x, aw = self.attention(x, hidden, enc_output) else: x, aw = x, 0 # passing the concatenated vector to the GRU # output: (batch_size, n_layers, hidden_size) # hidden: n_layers, batch_size, hidden_size | if LSTM (h_n, c_n) output, hidden = ( self.dec_rnn(x, hidden) if hidden is not None else self.dec_rnn(x) ) # output :shp: (batch_size * 1, hidden_size) output = output.view(-1, output.size(2)) # output :shp: (batch_size * 1, output_dim) output = self.fc(output) return output, hidden, aw class Seq2Seq(nn.Module): """ Class dependency: Encoder, Decoder """ def __init__( self, encoder, decoder, pass_enc2dec_hid=False, dropout=0, device="cpu" ): super(Seq2Seq, self).__init__() self.encoder = encoder self.decoder = decoder self.device = device self.pass_enc2dec_hid = pass_enc2dec_hid _force_en2dec_hid_conv = False if self.pass_enc2dec_hid: assert ( decoder.dec_hidden_dim == encoder.enc_hidden_dim ), "Hidden Dimension of encoder and decoder must be same, or unset `pass_enc2dec_hid`" if decoder.use_attention: assert ( decoder.enc_outstate_dim == encoder.enc_directions * encoder.enc_hidden_dim ), "Set `enc_out_dim` correctly in decoder" assert ( self.pass_enc2dec_hid or decoder.use_attention ), "No use of a decoder with No attention and No Hidden from Encoder" self.use_conv_4_enc2dec_hid = False if ( self.pass_enc2dec_hid and (encoder.enc_directions * encoder.enc_layers != decoder.dec_layers) ) or _force_en2dec_hid_conv: if encoder.enc_rnn_type == "lstm" or encoder.enc_rnn_type == "lstm": raise Exception( "XlitError: conv for enc2dec_hid not implemented; Change the layer numbers appropriately" ) self.use_conv_4_enc2dec_hid = True self.enc_hid_1ax = encoder.enc_directions * encoder.enc_layers self.dec_hid_1ax = decoder.dec_layers self.e2d_hidden_conv = nn.Conv1d(self.enc_hid_1ax, self.dec_hid_1ax, 1) def enc2dec_hidden(self, enc_hidden): """ enc_hidden: n_layer, batch_size, hidden_dim*num_directions TODO: Implement the logic for LSTm bsed model """ # hidden: batch_size, enc_layer*num_directions, enc_hidden_dim hidden = enc_hidden.permute(1, 0, 2).contiguous() # hidden: batch_size, dec_layers, dec_hidden_dim -> [N,C,Tstep] hidden = self.e2d_hidden_conv(hidden) # hidden: dec_layers, batch_size , dec_hidden_dim hidden_for_dec = hidden.permute(1, 0, 2).contiguous() return hidden_for_dec def active_beam_inference(self, src, beam_width=3, max_tgt_sz=50): """Search based decoding src: (sequence_len) """ def _avg_score(p_tup): """Used for Sorting TODO: Dividing by length of sequence power alpha as hyperparam """ return p_tup[0] import sys batch_size = 1 start_tok = src[0] end_tok = src[-1] src_sz = torch.tensor([len(src)]) src_ = src.unsqueeze(0) # enc_output: (batch_size, padded_seq_length, enc_hidden_dim*num_direction) # enc_hidden: (enc_layers*num_direction, batch_size, hidden_dim) enc_output, enc_hidden = self.encoder(src_, src_sz) if self.pass_enc2dec_hid: # dec_hidden: dec_layers, batch_size , dec_hidden_dim if self.use_conv_4_enc2dec_hid: init_dec_hidden = self.enc2dec_hidden(enc_hidden) else: init_dec_hidden = enc_hidden else: # dec_hidden -> Will be initialized to zeros internally init_dec_hidden = None # top_pred[][0] = Σ-log_softmax # top_pred[][1] = sequence torch.tensor shape: (1) # top_pred[][2] = dec_hidden top_pred_list = [(0, start_tok.unsqueeze(0), init_dec_hidden)] for t in range(max_tgt_sz): cur_pred_list = [] for p_tup in top_pred_list: if p_tup[1][-1] == end_tok: cur_pred_list.append(p_tup) continue # dec_hidden: dec_layers, 1, hidden_dim # dec_output: 1, output_dim dec_output, dec_hidden, _ = self.decoder( x=p_tup[1][-1].view(1, 1), # dec_input: (1,1) hidden=p_tup[2], enc_output=enc_output, ) ## π{prob} = Σ{log(prob)} -> to prevent diminishing # dec_output: (1, output_dim) dec_output = nn.functional.log_softmax(dec_output, dim=1) # pred_topk.values & pred_topk.indices: (1, beam_width) pred_topk = torch.topk(dec_output, k=beam_width, dim=1) for i in range(beam_width): sig_logsmx_ = p_tup[0] + pred_topk.values[0][i] # seq_tensor_ : (seq_len) seq_tensor_ = torch.cat((p_tup[1], pred_topk.indices[0][i].view(1))) cur_pred_list.append((sig_logsmx_, seq_tensor_, dec_hidden)) cur_pred_list.sort(key=_avg_score, reverse=True) # Maximized order top_pred_list = cur_pred_list[:beam_width] # check if end_tok of all topk end_flags_ = [1 if t[1][-1] == end_tok else 0 for t in top_pred_list] if beam_width == sum(end_flags_): break pred_tnsr_list = [t[1] for t in top_pred_list] return pred_tnsr_list ##===================== Glyph handlers ======================================= class GlyphStrawboss: def __init__(self, glyphs="en"): """list of letters in a language in unicode lang: ISO Language code glyphs: json file with script information """ if glyphs == "en": # Smallcase alone self.glyphs = [chr(alpha) for alpha in range(97, 122 + 1)] else: self.dossier = json.load(open(glyphs, encoding="utf-8")) self.glyphs = self.dossier["glyphs"] self.numsym_map = self.dossier["numsym_map"] self.char2idx = {} self.idx2char = {} self._create_index() def _create_index(self): self.char2idx["_"] = 0 # pad self.char2idx["$"] = 1 # start self.char2idx["#"] = 2 # end self.char2idx["*"] = 3 # Mask self.char2idx["'"] = 4 # apostrophe U+0027 self.char2idx["%"] = 5 # unused self.char2idx["!"] = 6 # unused # letter to index mapping for idx, char in enumerate(self.glyphs): self.char2idx[char] = idx + 7 # +7 token initially # index to letter mapping for char, idx in self.char2idx.items(): self.idx2char[idx] = char def size(self): return len(self.char2idx) def word2xlitvec(self, word): """Converts given string of gyphs(word) to vector(numpy) Also adds tokens for start and end """ try: vec = [self.char2idx["$"]] # start token for i in list(word): vec.append(self.char2idx[i]) vec.append(self.char2idx["#"]) # end token vec = np.asarray(vec, dtype=np.int64) return vec except Exception as error: print("XlitError: In word:", word, "Error Char not in Token:", error) sys.exit() def xlitvec2word(self, vector): """Converts vector(numpy) to string of glyphs(word)""" char_list = [] for i in vector: char_list.append(self.idx2char[i]) word = "".join(char_list).replace("$", "").replace("#", "") # remove tokens word = word.replace("_", "").replace("*", "") # remove tokens return word class VocabSanitizer: def __init__(self, data_file): """ data_file: path to file conatining vocabulary list """ extension = os.path.splitext(data_file)[-1] if extension == ".json": self.vocab_set = set(json.load(open(data_file, encoding="utf-8"))) elif extension == ".csv": self.vocab_df = pd.read_csv(data_file).set_index("WORD") self.vocab_set = set(self.vocab_df.index) else: print("XlitError: Only Json/CSV file extension supported") def reposition(self, word_list): """Reorder Words in list""" new_list = [] temp_ = word_list.copy() for v in word_list: if v in self.vocab_set: new_list.append(v) temp_.remove(v) new_list.extend(temp_) return new_list ##=============== INSTANTIATION ================================================ class XlitPiston: """ For handling prediction & post-processing of transliteration for a single language Class dependency: Seq2Seq, GlyphStrawboss, VocabSanitizer Global Variables: F_DIR """ def __init__( self, weight_path, vocab_file, tglyph_cfg_file, iglyph_cfg_file="en", device="cpu", ): self.device = device self.in_glyph_obj = GlyphStrawboss(iglyph_cfg_file) self.tgt_glyph_obj = GlyphStrawboss(glyphs=tglyph_cfg_file) self.voc_sanity = VocabSanitizer(vocab_file) self._numsym_set = set( json.load(open(tglyph_cfg_file, encoding="utf-8"))["numsym_map"].keys() ) self._inchar_set = set("abcdefghijklmnopqrstuvwxyz") self._natscr_set = set().union( self.tgt_glyph_obj.glyphs, sum(self.tgt_glyph_obj.numsym_map.values(), []) ) ## Model Config Static TODO: add defining in json support input_dim = self.in_glyph_obj.size() output_dim = self.tgt_glyph_obj.size() enc_emb_dim = 300 dec_emb_dim = 300 enc_hidden_dim = 512 dec_hidden_dim = 512 rnn_type = "lstm" enc2dec_hid = True attention = True enc_layers = 1 dec_layers = 2 m_dropout = 0 enc_bidirect = True enc_outstate_dim = enc_hidden_dim * (2 if enc_bidirect else 1) enc = Encoder( input_dim=input_dim, embed_dim=enc_emb_dim, hidden_dim=enc_hidden_dim, rnn_type=rnn_type, layers=enc_layers, dropout=m_dropout, device=self.device, bidirectional=enc_bidirect, ) dec = Decoder( output_dim=output_dim, embed_dim=dec_emb_dim, hidden_dim=dec_hidden_dim, rnn_type=rnn_type, layers=dec_layers, dropout=m_dropout, use_attention=attention, enc_outstate_dim=enc_outstate_dim, device=self.device, ) self.model = Seq2Seq(enc, dec, pass_enc2dec_hid=enc2dec_hid, device=self.device) self.model = self.model.to(self.device) weights = torch.load(weight_path, map_location=torch.device(self.device)) self.model.load_state_dict(weights) self.model.eval() def character_model(self, word, beam_width=1): in_vec = torch.from_numpy(self.in_glyph_obj.word2xlitvec(word)).to(self.device) ## change to active or passive beam p_out_list = self.model.active_beam_inference(in_vec, beam_width=beam_width) p_result = [ self.tgt_glyph_obj.xlitvec2word(out.cpu().numpy()) for out in p_out_list ] result = self.voc_sanity.reposition(p_result) # List type return result def numsym_model(self, seg): """tgt_glyph_obj.numsym_map[x] returns a list object""" if len(seg) == 1: return [seg] + self.tgt_glyph_obj.numsym_map[seg] a = [self.tgt_glyph_obj.numsym_map[n][0] for n in seg] return [seg] + ["".join(a)] def _word_segementer(self, sequence): sequence = sequence.lower() accepted = set().union(self._numsym_set, self._inchar_set, self._natscr_set) # sequence = ''.join([i for i in sequence if i in accepted]) segment = [] idx = 0 seq_ = list(sequence) while len(seq_): # for Number-Symbol temp = "" while len(seq_) and seq_[0] in self._numsym_set: temp += seq_[0] seq_.pop(0) if temp != "": segment.append(temp) # for Target Chars temp = "" while len(seq_) and seq_[0] in self._natscr_set: temp += seq_[0] seq_.pop(0) if temp != "": segment.append(temp) # for Input-Roman Chars temp = "" while len(seq_) and seq_[0] in self._inchar_set: temp += seq_[0] seq_.pop(0) if temp != "": segment.append(temp) temp = "" while len(seq_) and seq_[0] not in accepted: temp += seq_[0] seq_.pop(0) if temp != "": segment.append(temp) return segment def inferencer(self, sequence, beam_width=10): seg = self._word_segementer(sequence[:120]) lit_seg = [] p = 0 while p < len(seg): if seg[p][0] in self._natscr_set: lit_seg.append([seg[p]]) p += 1 elif seg[p][0] in self._inchar_set: lit_seg.append(self.character_model(seg[p], beam_width=beam_width)) p += 1 elif seg[p][0] in self._numsym_set: # num & punc lit_seg.append(self.numsym_model(seg[p])) p += 1 else: lit_seg.append([seg[p]]) p += 1 ## IF segment less/equal to 2 then return combinotorial, ## ELSE only return top1 of each result concatenated if len(lit_seg) == 1: final_result = lit_seg[0] elif len(lit_seg) == 2: final_result = [""] for seg in lit_seg: new_result = [] for s in seg: for f in final_result: new_result.append(f + s) final_result = new_result else: new_result = [] for seg in lit_seg: new_result.append(seg[0]) final_result = ["".join(new_result)] return final_result from collections.abc import Iterable from pydload import dload import zipfile MODEL_DOWNLOAD_URL_PREFIX = "https://github.com/AI4Bharat/IndianNLP-Transliteration/releases/download/xlit_v0.5.0/" def is_folder_writable(folder): try: os.makedirs(folder, exist_ok=True) tmp_file = os.path.join(folder, ".write_test") with open(tmp_file, "w") as f: f.write("Permission Check") os.remove(tmp_file) return True except: return False def is_directory_writable(path): if os.name == "nt": return is_folder_writable(path) return os.access(path, os.W_OK | os.X_OK) class XlitEngine: """ For Managing the top level tasks and applications of transliteration Global Variables: F_DIR """ def __init__( self, lang2use="all", config_path="translit_models/default_lineup.json" ): lineup = json.load(open(os.path.join(F_DIR, config_path), encoding="utf-8")) self.lang_config = {} if isinstance(lang2use, str): if lang2use == "all": self.lang_config = lineup elif lang2use in lineup: self.lang_config[lang2use] = lineup[lang2use] else: raise Exception( "XlitError: The entered Langauge code not found. Available are {}".format( lineup.keys() ) ) elif isinstance(lang2use, Iterable): for l in lang2use: try: self.lang_config[l] = lineup[l] except: print( "XlitError: Language code {} not found, Skipping...".format(l) ) else: raise Exception( "XlitError: lang2use must be a list of language codes (or) string of single language code" ) if is_directory_writable(F_DIR): models_path = os.path.join(F_DIR, "translit_models") else: user_home = os.path.expanduser("~") models_path = os.path.join(user_home, ".AI4Bharat_Xlit_Models") os.makedirs(models_path, exist_ok=True) self.download_models(models_path) self.langs = {} self.lang_model = {} for la in self.lang_config: try: print("Loading {}...".format(la)) self.lang_model[la] = XlitPiston( weight_path=os.path.join( models_path, self.lang_config[la]["weight"] ), vocab_file=os.path.join(models_path, self.lang_config[la]["vocab"]), tglyph_cfg_file=os.path.join( models_path, self.lang_config[la]["script"] ), iglyph_cfg_file="en", ) self.langs[la] = self.lang_config[la]["name"] except Exception as error: print("XlitError: Failure in loading {} \n".format(la), error) print(XlitError.loading_err.value) def download_models(self, models_path): """ Download models from GitHub Releases if not exists """ for l in self.lang_config: lang_name = self.lang_config[l]["eng_name"] lang_model_path = os.path.join(models_path, lang_name) if not os.path.isdir(lang_model_path): print("Downloading model for language: %s" % lang_name) remote_url = MODEL_DOWNLOAD_URL_PREFIX + lang_name + ".zip" downloaded_zip_path = os.path.join(models_path, lang_name + ".zip") dload(url=remote_url, save_to_path=downloaded_zip_path, max_time=None) if not os.path.isfile(downloaded_zip_path): exit( f"ERROR: Unable to download model from {remote_url} into {models_path}" ) with zipfile.ZipFile(downloaded_zip_path, "r") as zip_ref: zip_ref.extractall(models_path) if os.path.isdir(lang_model_path): os.remove(downloaded_zip_path) else: exit( f"ERROR: Unable to find models in {lang_model_path} after download" ) return def translit_word(self, eng_word, lang_code="default", topk=7, beam_width=10): if eng_word == "": return [] if lang_code in self.langs: try: res_list = self.lang_model[lang_code].inferencer( eng_word, beam_width=beam_width ) return res_list[:topk] except Exception as error: print("XlitError:", traceback.format_exc()) print(XlitError.internal_err.value) return XlitError.internal_err elif lang_code == "default": try: res_dict = {} for la in self.lang_model: res = self.lang_model[la].inferencer( eng_word, beam_width=beam_width ) res_dict[la] = res[:topk] return res_dict except Exception as error: print("XlitError:", traceback.format_exc()) print(XlitError.internal_err.value) return XlitError.internal_err else: print("XlitError: Unknown Langauge requested", lang_code) print(XlitError.lang_err.value) return XlitError.lang_err def translit_sentence(self, eng_sentence, lang_code="default", beam_width=10): if eng_sentence == "": return [] if lang_code in self.langs: try: out_str = "" for word in eng_sentence.split(): res_ = self.lang_model[lang_code].inferencer( word, beam_width=beam_width ) out_str = out_str + res_[0] + " " return out_str[:-1] except Exception as error: print("XlitError:", traceback.format_exc()) print(XlitError.internal_err.value) return XlitError.internal_err elif lang_code == "default": try: res_dict = {} for la in self.lang_model: out_str = "" for word in eng_sentence.split(): res_ = self.lang_model[la].inferencer( word, beam_width=beam_width ) out_str = out_str + res_[0] + " " res_dict[la] = out_str[:-1] return res_dict except Exception as error: print("XlitError:", traceback.format_exc()) print(XlitError.internal_err.value) return XlitError.internal_err else: print("XlitError: Unknown Langauge requested", lang_code) print(XlitError.lang_err.value) return XlitError.lang_err if __name__ == "__main__": available_lang = [ "bn", "gu", "hi", "kn", "gom", "mai", "ml", "mr", "pa", "sd", "si", "ta", "te", "ur", ] reg = re.compile(r"[a-zA-Z]") lang = "hi" engine = XlitEngine( lang ) # if you don't specify lang code here, this will give results in all langs available sent = "Hello World! ABCD क्या हाल है आपका?" words = [ engine.translit_word(word, topk=1)[lang][0] if reg.match(word) else word for word in sent.split() ] # only transliterated en words, leaves rest as it is updated_sent = " ".join(words) print(updated_sent) # output : हेलो वर्ल्ड! क्या हाल है आपका? # y = engine.translit_sentence("Hello World !")['hi'] # print(y)