| INDIC_NLP_LIB_HOME = "indic_nlp_library" |
| INDIC_NLP_RESOURCES = "indic_nlp_resources" |
| import sys |
|
|
| from indicnlp import transliterate |
|
|
| sys.path.append(r"{}".format(INDIC_NLP_LIB_HOME)) |
| from indicnlp import common |
|
|
| common.set_resources_path(INDIC_NLP_RESOURCES) |
| from indicnlp import loader |
|
|
| loader.load() |
| from sacremoses import MosesPunctNormalizer |
| from sacremoses import MosesTokenizer |
| from sacremoses import MosesDetokenizer |
| from collections import defaultdict |
|
|
| import indicnlp |
| from indicnlp.tokenize import indic_tokenize |
| from indicnlp.tokenize import indic_detokenize |
| from indicnlp.normalize import indic_normalize |
| from indicnlp.transliterate import unicode_transliterate |
|
|
| from flores_codes_map_indic import flores_codes |
| import sentencepiece as spm |
|
|
| import re |
|
|
| en_detok = MosesDetokenizer(lang="en") |
|
|
|
|
| def postprocess( |
| infname: str, |
| outfname: str, |
| input_size: int, |
| lang: str, |
| transliterate: bool = False, |
| spm_model_path: str = None, |
| ): |
| """ |
| Postprocess the output of a machine translation model in the following order: |
| - parse fairseq interactive output |
| - convert script back to native Indic script (in case of Indic languages) |
| - detokenize |
| |
| Args: |
| infname (str): path to the input file containing the machine translation output. |
| outfname (str): path to the output file where the postprocessed output will be written. |
| input_size (int): number of sentences in the input file. |
| lang (str): language code of the output language. |
| transliterate (bool, optional): whether to transliterate the output text to devanagari (default: False). |
| spm_model_path (str): path of the sentence piece model. |
| """ |
| if spm_model_path is None: |
| raise Exception("Please provide sentence piece model path for decoding") |
| |
| sp = spm.SentencePieceProcessor(model_file=spm_model_path) |
| |
| iso_lang = flores_codes[lang] |
|
|
| consolidated_testoutput = [] |
| consolidated_testoutput = [(x, 0.0, "") for x in range(input_size)] |
|
|
| temp_testoutput = [] |
| with open(infname, "r", encoding="utf-8") as infile: |
| temp_testoutput = list( |
| map( |
| lambda x: x.strip().split("\t"), |
| filter(lambda x: x.startswith("H-"), infile), |
| ) |
| ) |
| temp_testoutput = list( |
| map(lambda x: (int(x[0].split("-")[1]), float(x[1]), x[2]), temp_testoutput) |
| ) |
| for sid, score, hyp in temp_testoutput: |
| consolidated_testoutput[sid] = (sid, score, hyp) |
| consolidated_testoutput = [x[2] for x in consolidated_testoutput] |
| consolidated_testoutput = [sp.decode(x.split(" ")) for x in consolidated_testoutput] |
|
|
| if iso_lang == "en": |
| with open(outfname, "w", encoding="utf-8") as outfile: |
| for sent in consolidated_testoutput: |
| outfile.write(en_detok.detokenize(sent.split(" ")) + "\n") |
| else: |
| xliterator = unicode_transliterate.UnicodeIndicTransliterator() |
| with open(outfname, "w", encoding="utf-8") as outfile: |
| for sent in consolidated_testoutput: |
| if transliterate: |
| outstr = indic_detokenize.trivial_detokenize( |
| xliterator.transliterate(sent, "hi", iso_lang), iso_lang |
| ) |
| else: |
| outstr = indic_detokenize.trivial_detokenize(sent, iso_lang) |
| outfile.write(outstr + "\n") |
|
|
|
|
| if __name__ == "__main__": |
| infname = sys.argv[1] |
| outfname = sys.argv[2] |
| input_size = int(sys.argv[3]) |
| lang = sys.argv[4] |
| transliterate = sys.argv[5] |
| spm_model_path = sys.argv[6] |
|
|
| postprocess(infname, outfname, input_size, lang, transliterate, spm_model_path) |
|
|
