app.py

import gradio as gr
import spaces
import torch
from transformers import T5Tokenizer, T5ForConditionalGeneration, AutoTokenizer, AutoModelForSeq2SeqLM, AutoModelForCausalLM, AutoModel, pipeline
from transformers import logging as hflogging
import languagecodes
import httpx, os
import polars as pl 

hflogging.set_verbosity_error()
favourite_langs = {"German": "de", "Romanian": "ro", "English": "en", "-----": "-----"}
df = pl.read_parquet("isolanguages.parquet")
non_empty_isos = df.slice(1).filter(pl.col("ISO639-1") != "").rows()
# all_langs = languagecodes.iso_languages_byname
all_langs = {iso[0]: (iso[1], iso[2], iso[3]) for iso in non_empty_isos} # {'Romanian': ('ro', 'rum', 'ron')}
iso1toall = {iso[1]: (iso[0], iso[2], iso[3]) for iso in non_empty_isos} # {'ro': ('Romanian', 'rum', 'ron')}
langs = list(favourite_langs.keys())
langs.extend(list(all_langs.keys())) # Language options as list, add favourite languages first

models = ["Helsinki-NLP", "QUICKMT", "Argos", "HPLT", "HPLT-OPUS", "Google",
          "Helsinki-NLP/opus-mt-tc-bible-big-mul-mul", "Helsinki-NLP/opus-mt-tc-bible-big-mul-deu_eng_nld",
          "Helsinki-NLP/opus-mt-tc-bible-big-mul-deu_eng_fra_por_spa", "Helsinki-NLP/opus-mt-tc-bible-big-deu_eng_fra_por_spa-mul",
          "Helsinki-NLP/opus-mt-tc-bible-big-roa-deu_eng_fra_por_spa", "Helsinki-NLP/opus-mt-tc-bible-big-deu_eng_fra_por_spa-roa", "Helsinki-NLP/opus-mt-tc-bible-big-roa-en",
          "facebook/nllb-200-distilled-600M", "facebook/nllb-200-distilled-1.3B", "facebook/nllb-200-1.3B", "facebook/nllb-200-3.3B",
          "facebook/mbart-large-50-many-to-many-mmt", "facebook/mbart-large-50-one-to-many-mmt", "facebook/mbart-large-50-many-to-one-mmt",
          "facebook/hf-seamless-m4t-medium", "facebook/seamless-m4t-large", "facebook/seamless-m4t-v2-large",
          "facebook/m2m100_418M", "facebook/m2m100_1.2B",
          "alirezamsh/small100", "naist-nlp/mitre_466m", "naist-nlp/mitre_913m",
          "bigscience/mt0-small", "bigscience/mt0-base", "bigscience/mt0-large", "bigscience/mt0-xl",
          "bigscience/bloomz-560m", "bigscience/bloomz-1b1", "bigscience/bloomz-1b7", "bigscience/bloomz-3b",
          "google/madlad400-3b-mt", "jbochi/madlad400-3b-mt",
          "NiuTrans/LMT-60-0.6B", "NiuTrans/LMT-60-1.7B", "NiuTrans/LMT-60-4B",
          "Lego-MT/Lego-MT", "BSC-LT/salamandraTA-2b-instruct",
          "winninghealth/WiNGPT-Babel", "winninghealth/WiNGPT-Babel-2", "winninghealth/WiNGPT-Babel-2.1",
          "Unbabel/Tower-Plus-2B", "utter-project/EuroLLM-1.7B", "utter-project/EuroLLM-1.7B-Instruct",
          "yanolja/YanoljaNEXT-Rosetta-4B-2511", "yanolja/YanoljaNEXT-Rosetta-4B",
          "google-t5/t5-small", "google-t5/t5-base", "google-t5/t5-large",
          "google/flan-t5-small", "google/flan-t5-base", "google/flan-t5-large", "google/flan-t5-xl"]
DEFAULTS = [langs[0], langs[1], models[0]]

def timer(func):
    from time import time
    def translate(input_text, s_language, t_language, model_name) -> tuple[str, str]:
        start_time = time()
        translated_text, message_text = func(input_text, s_language, t_language, model_name)
        end_time = time()
        execution_time = end_time - start_time
        # print(f"Function {func.__name__!r} executed in {execution_time:.2f} seconds.")
        message_text = f'Executed in {execution_time:.2f} seconds! {message_text}' 
        return translated_text, message_text
    return translate

def model_to_cuda(model):
    # Move the model to GPU if available
    if torch.cuda.is_available():
        model = model.to('cuda')
        print("CUDA is available! Using GPU.")
    else:
        print("CUDA not available! Using CPU.")
    return model

def HelsinkiNLPAutoTokenizer(sl, tl, input_text): # deprecated
    if model_name == "Helsinki-NLP":
        message_text = f'Translated from {sl} to {tl} with {model_name}.'
        try:
            model_name = f"Helsinki-NLP/opus-mt-{sl}-{tl}"
            tokenizer = AutoTokenizer.from_pretrained(model_name)
            model = model_to_cuda(AutoModelForSeq2SeqLM.from_pretrained(model_name))
        except EnvironmentError:
            try:   
                model_name = f"Helsinki-NLP/opus-tatoeba-{sl}-{tl}"
                tokenizer = AutoTokenizer.from_pretrained(model_name)
                model = model_to_cuda(AutoModelForSeq2SeqLM.from_pretrained(model_name))
                input_ids = tokenizer.encode(prompt, return_tensors="pt")
                output_ids = model.generate(input_ids, max_length=512)
                translated_text = tokenizer.decode(output_ids[0], skip_special_tokens=True)
                return translated_text, message_text
            except EnvironmentError as error:
                return f"Error finding model: {model_name}! Try other available language combination.", error
                      
class Translators:
    def __init__(self, model_name: str, sl: str, tl: str, input_text: str):
        self.model_name = model_name
        self.sl, self.tl = sl, tl
        self.input_text = input_text
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        self.max_new_tokens = 512
        
    def google(self):  
        self.input_text = " ".join(self.input_text.split())
        url = os.environ['GCLIENT'] + f'sl={self.sl}&tl={self.tl}&q={self.input_text}'
        response = httpx.get(url)
        return response.json()[0][0][0]
    
    def simplepipe(self):
        try:
            pipe = pipeline("translation", model=self.model_name, device=self.device)                
            translation = pipe(self.input_text)
            message = f'Translated from {iso1toall[self.sl][0]} to {iso1toall[self.tl][0]} with {self.model_name}.'
            return translation[0]['translation_text'], message
        except Exception as error:
            return f"Error translating with model: {self.model_name}! Try other available language combination or model.", error

    def mitre(self):
        from transformers import AutoModel, AutoTokenizer
        tokenizer = AutoTokenizer.from_pretrained(self.model_name, trust_remote_code=True, use_fast=False)
        model = AutoModel.from_pretrained(self.model_name, trust_remote_code=True).to(self.device)
        model.eval()
        # Translating from one or several sentences to a sole language
        src_tokens = tokenizer.encode_source_tokens_to_input_ids(self.input_text, target_language=self.tl)
        with torch.inference_mode(): # no_grad inference_mode
            generated_tokens = model.generate(src_tokens)
        result = tokenizer.batch_decode(generated_tokens, skip_special_tokens=True)[0]
        return result
    
    def rosetta(self):
        model = AutoModelForCausalLM.from_pretrained(
            self.model_name,
            dtype=torch.bfloat16, # float32 slow
            low_cpu_mem_usage=False, # True
            device_map="auto")
        tokenizer = AutoTokenizer.from_pretrained(self.model_name)
        system = f"Translate the user's text to {self.tl}. Provide the final translation in a formal tone immediately immediately without any other text."
        messages = [
            {"role": "system", "content": system},
            {"role": "user", "content": self.input_text},
        ]     
        prompt = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
        inputs = tokenizer(prompt, return_tensors="pt").to(self.device)
        input_length = inputs["input_ids"].shape[1]       
        model.eval()
        with torch.inference_mode():
            outputs = model.generate(
                **inputs,
                max_new_tokens=self.max_new_tokens,
            )   
        generated_tokens = outputs[0][input_length:]
        translation = tokenizer.decode(generated_tokens, skip_special_tokens=True)
        return translation    
    
    def niutrans(self):
        tokenizer = AutoTokenizer.from_pretrained(self.model_name, padding_side='left')
        model = AutoModelForCausalLM.from_pretrained(self.model_name)
        prompt = f"Translate the following text from {self.sl} into {self.tl}.\n{self.sl}: {self.input_text}.\n{self.tl}: "
        messages = [{"role": "user", "content": prompt}]
        text = tokenizer.apply_chat_template(
            messages,
            tokenize=False,
            add_generation_prompt=True,
        )
        model_inputs = tokenizer([text], return_tensors="pt").to(model.device)
        generated_ids = model.generate(**model_inputs, max_new_tokens=512, num_beams=5, do_sample=False)
        output_ids = generated_ids[0][len(model_inputs.input_ids[0]):].tolist() 
        outputs = tokenizer.batch_decode(output_ids, skip_special_tokens=True)
        outputs = ''.join(outputs) if isinstance(outputs, list) else outputs
        return outputs

    def salamandratapipe(self):
        pipe = pipeline("text-generation", model=self.model_name)
        messages = [{"role": "user", "content": f"Translate the following text from {self.sl} into {self.tl}.\n{self.sl}: {self.input_text} \n{self.tl}:"}]
        return pipe(messages, max_new_tokens=self.max_new_tokens, early_stopping=True, num_beams=5)[0]["generated_text"][1]["content"]
     
    def hplt(self, opus = False):
        # langs = ['ar', 'bs', 'ca', 'en', 'et', 'eu', 'fi', 'ga', 'gl', 'hi', 'hr', 'is', 'mt', 'nn', 'sq', 'sw', 'zh_hant']
        hplt_models = ['ar-en', 'bs-en', 'ca-en', 'en-ar', 'en-bs', 'en-ca', 'en-et', 'en-eu', 'en-fi',
                  'en-ga', 'en-gl', 'en-hi', 'en-hr', 'en-is', 'en-mt', 'en-nn', 'en-sq', 'en-sw',
                  'en-zh_hant', 'et-en', 'eu-en', 'fi-en', 'ga-en', 'gl-en', 'hi-en', 'hr-en',
                  'is-en', 'mt-en', 'nn-en', 'sq-en', 'sw-en', 'zh_hant-en']
        lang_map = {"zh": "zh_hant"}
        self.sl = lang_map.get(self.sl, self.sl)
        self.tl = lang_map.get(self.tl, self.tl)
        if opus:
            hplt_model = f'HPLT/translate-{self.sl}-{self.tl}-v1.0-hplt_opus' # HPLT/translate-en-hr-v1.0-hplt_opus
        else:
            hplt_model = f'HPLT/translate-{self.sl}-{self.tl}-v1.0-hplt' # HPLT/translate-en-hr-v1.0-hplt
        if f'{self.sl}-{self.tl}' in hplt_models:
            pipe = pipeline("translation", model=hplt_model, device=self.device)
            translation = pipe(self.input_text)
            translated_text = translation[0]['translation_text']
            message_text = f'Translated from {iso1toall[self.sl][0]} to {iso1toall[self.tl][0]} with {hplt_model}.'
        else:
            translated_text = f'HPLT model from {iso1toall[self.sl][0]} to {iso1toall[self.tl][0]} not available!'
            message_text = f"Available models: {', '.join(hplt_models)}"
        return translated_text, message_text

    @staticmethod
    def download_argos_model(available_packages, from_code, to_code):
        import argostranslate.package
        print('Downloading model for', from_code, to_code) 
        # Download and install Argos Translate package from path
        package_to_install = next(
            filter(lambda x: x.from_code == from_code and x.to_code == to_code, available_packages)
        )
        argostranslate.package.install_from_path(package_to_install.download())
    
    def argos(self):
        import argostranslate.translate, argostranslate.package
        argostranslate.package.update_package_index()
        available_packages = argostranslate.package.get_available_packages()
        available_slanguages = [lang.from_code for lang in available_packages]
        available_tlanguages = [lang.to_code for lang in available_packages]
        available_languages = sorted(list(set(available_slanguages + available_tlanguages)))
        combos: tuple[str|str] = sorted(list(zip(available_slanguages, available_tlanguages)))
        packages_info = ', '.join(f"{pkg.from_name} ({pkg.from_code}) -> {pkg.to_name} ({pkg.to_code})" for pkg in available_packages)
        # print(available_languages, combos, packages_info)
        if self.sl not in available_languages and self.tl not in available_languages:
            translated_text = f'''No supported Argos model available from {iso1toall[self.sl][0]} to {iso1toall[self.tl][0]}!
            Try other model or languages combination from the available Argos models: {', '.join(available_languages)}.'''
        else:
            try:
                if (self.sl, self.tl) in combos:
                    self.__class__.download_argos_model(available_packages, self.sl, self.tl) # Download model
                    translated_text = argostranslate.translate.translate(self.input_text, self.sl, self.tl) # Direct translation
                elif (self.sl, 'en') in combos and ('en', self.tl) in combos:
                    self.__class__.download_argos_model(available_packages, self.sl, 'en') # Download model
                    translated_pivottext = argostranslate.translate.translate(self.input_text, self.sl, 'en') # Translate to pivot language English
                    self.__class__.download_argos_model(available_packages, 'en', self.tl) # Download model
                    translated_text = argostranslate.translate.translate(translated_pivottext, 'en', self.tl) # Translate from pivot language English
                    message = f'Translated from {iso1toall[self.sl][0]} to {iso1toall[self.tl][0]} with Argos using pivot language English.'
                else:
                    translated_text = f"No Argos model for {iso1toall[self.sl][0]} to {iso1toall[self.tl][0]}. Try other model or languages combination from the available Argos models: {packages_info}."
            except StopIteration as IterationError:
                # packages_info = ', '.join(f"{pkg.get_description()}->{str(pkg.links)} {str(pkg.source_languages)}" for pkg in available_packages)
                translated_text = f"No Argos model for {iso1toall[self.sl][0]} to {iso1toall[self.tl][0]}. Error: {IterationError}. Try other model or languages combination from the available Argos models: {packages_info}."
            except Exception as generalerror:
                translated_text = f"General error: {generalerror}"
        return translated_text

    @staticmethod
    def quickmttranslate(model_path, input_text):
        from quickmt import Translator
        # 'auto' auto-detects GPU, set to "cpu" to force CPU inference
        # device = 'gpu' if torch.cuda.is_available() else 'cpu'
        translator = Translator(str(model_path), device="auto", compute_type="auto")       
        # translation = Translator(f"./quickmt-{self.sl}-{self.tl}/", device="auto/cpu", intra_threads=2, inter_threads=2, compute_type="int8")       
        # ctranslate2._ext.Translator(model_path: str, device: str = 'cpu', *, device_index: Union[int, List[int]] = 0, compute_type: Union[str, Dict[str, str]] = 'default',
        # inter_threads: int = 1, intra_threads: int = 0, max_queued_batches: int = 0, flash_attention: bool = False, tensor_parallel: bool = False, files: object = None)
        # Options for compute_type: default, auto, int8, int8_float32, int8_float16, int8_bfloat16, int16, float16, bfloat16, float32
        # "int8" will work well for inference on CPU and give "int8_float16" or "int8_bfloat16" a try for GPU inference.
        
        # (self: ctranslate2._ext.Translator, source: List[List[str]], target_prefix: Optional[List[Optional[List[str]]]] = None, *, max_batch_size: int = 0,
        # batch_type: str = 'examples', asynchronous: bool = False, beam_size: int = 2, patience: float = 1, num_hypotheses: int = 1, length_penalty: float = 1,
        # coverage_penalty: float = 0, repetition_penalty: float = 1, no_repeat_ngram_size: int = 0, disable_unk: bool = False,
        # suppress_sequences: Optional[List[List[str]]] = None, end_token: Optional[Union[str, List[str], List[int]]] = None, return_end_token: bool = False,
        # prefix_bias_beta: float = 0, max_input_length: int = 1024, max_decoding_length: int = 256, min_decoding_length: int = 1, use_vmap: bool = False,
        # return_scores: bool = False, return_logits_vocab: bool = False, return_attention: bool = False, return_alternatives: bool = False,
        # min_alternative_expansion_prob: float = 0, sampling_topk: int = 1, sampling_topp: float = 1, sampling_temperature: float = 1, replace_unknowns: bool = False,
        # callback: Callable[[ctranslate2._ext.GenerationStepResult], bool] = None) -> Union[List[ctranslate2._ext.TranslationResult], List[ctranslate2._ext.AsyncTranslationResult]]
        # set beam size to 1 for faster speed (but lower quality) device="auto/cpu/gpu"
        translation = translator(input_text, beam_size=5, max_input_length = 512, max_decoding_length = 512)
        # print(model_path, input_text, translation)
        return translation

    @staticmethod
    def quickmtdownload(model_name):
        from quickmt.hub import hf_download
        from pathlib import Path
        model_path = Path("/quickmt/models") / model_name
        if not model_path.exists():
            hf_download(
            model_name = f"quickmt/{model_name}",
            output_dir=Path("/quickmt/models") / model_name,
        )
        return model_path
            
    def quickmt(self):
        model_name = f"quickmt-{self.sl}-{self.tl}"
        # from quickmt.hub import hf_list
        # quickmt_models = [i.split("/quickmt-")[1] for i in hf_list()]
        # quickmt_models.sort()
        quickmt_models = ['ar-en', 'bn-en', 'cs-en', 'da-en', 'de-en', 'el-en', 'en-ar', 'en-bn',
                          'en-cs', 'en-da', 'en-de', 'en-el', 'en-es', 'en-fa', 'en-fr', 'en-he',
                          'en-hi', 'en-hu', 'en-id', 'en-is', 'en-it', 'en-ja', 'en-ko', 'en-lv', 'en-pl',
                          'en-pt', 'en-ro', 'en-ru', 'en-sv', 'en-th', 'en-tr', 'en-ur', 'en-vi',
                          'en-zh', 'es-en', 'fa-en', 'fr-en', 'he-en', 'hi-en', 'hu-en', 'id-en',
                          'is-en', 'it-en', 'ja-en', 'ko-en', 'lv-en', 'pl-en', 'pt-en', 'ro-en', 'ru-en',
                          'th-en', 'tr-en', 'ur-en', 'vi-en', 'zh-en']
        # available_languages = list(set([lang for model in quickmt_models for lang in model.split('-')]))
        # available_languages.sort()
        available_languages = ['ar', 'bn', 'cs', 'da', 'de', 'el', 'en', 'es', 'fa', 'fr', 'he',
                               'hi', 'hu', 'id', 'it', 'is', 'ja', 'ko', 'lv', 'pl', 'pt', 'ro', 'ru',
                               'sv', 'th', 'tr', 'ur', 'vi', 'zh']
        # print(quickmt_models, available_languages)       
        # Direct translation model
        if f"{self.sl}-{self.tl}" in quickmt_models:
            model_path = Translators.quickmtdownload(model_name)
            translated_text = Translators.quickmttranslate(model_path, self.input_text)
            message = f'Translated from {iso1toall[self.sl][0]} to {iso1toall[self.tl][0]} with {model_name}.'
        # Pivot language English
        elif self.sl in available_languages and self.tl in available_languages:
            model_name = f"quickmt-{self.sl}-en"
            model_path = Translators.quickmtdownload(model_name)
            entranslation = Translators.quickmttranslate(model_path, self.input_text)
            model_name = f"quickmt-en-{self.tl}"
            model_path = Translators.quickmtdownload(model_name)
            translated_text = Translators.quickmttranslate(model_path, entranslation)
            message = f'Translated from {iso1toall[self.sl][0]} to {iso1toall[self.tl][0]} with Quickmt using pivot language English.'
        else:
            translated_text = f'No Quickmt model available for translation from {iso1toall[self.sl][0]} to {iso1toall[self.tl][0]}!'
            message = f"Available models: {', '.join(quickmt_models)}"
        return translated_text, message
        
    def HelsinkiNLP_mulroa(self):
        try:
            pipe = pipeline("translation", model=self.model_name, device=self.device)                
            tgt_lang = iso1toall.get(self.tl)[2] # 'deu', 'ron', 'eng', 'fra'
            translation = pipe(f'>>{tgt_lang}<< {self.input_text}')
            return translation[0]['translation_text'], f'Translated from {iso1toall[self.sl][0]} to {iso1toall[self.tl][0]} with {self.model_name}.'
        except Exception as error:
            return f"Error translating with model: {self.model_name}! Try other available language combination.", error
    
    def HelsinkiNLP(self):
        try: # Standard bilingual model
            model_name = f"Helsinki-NLP/opus-mt-{self.sl}-{self.tl}"
            pipe = pipeline("translation", model=model_name, device=self.device)
            translation = pipe(self.input_text)
            return translation[0]['translation_text'], f'Translated from {iso1toall[self.sl][0]} to {iso1toall[self.tl][0]} with {model_name}.'
        except EnvironmentError:
            try: # Tatoeba models
                model_name = f"Helsinki-NLP/opus-tatoeba-{self.sl}-{self.tl}"
                pipe = pipeline("translation", model=model_name, device=self.device)
                translation = pipe(self.input_text)
                return translation[0]['translation_text'], f'Translated from {iso1toall[self.sl][0]} to {iso1toall[self.tl][0]} with {model_name}.'
            except EnvironmentError as error:
                self.model_name = "Helsinki-NLP/opus-mt-tc-bible-big-mul-mul" # Last resort: try multi to multi
                return self.HelsinkiNLP_mulroa()
        except KeyError as error:
            return f"Error: Translation direction {self.sl} to {self.tl} is not supported by Helsinki Translation Models", error
     
    def madlad(self):
        model = T5ForConditionalGeneration.from_pretrained(self.model_name, device_map="auto")
        tokenizer = T5Tokenizer.from_pretrained(self.model_name)
        text = f"<2{self.tl}> {self.input_text}"
        # input_ids = tokenizer(text, return_tensors="pt").input_ids.to(model.device)
        # outputs = model.generate(input_ids=input_ids, max_new_tokens=512)    
        # return tokenizer.decode(outputs[0], skip_special_tokens=True)
        # return tokenizer.batch_decode(outputs, skip_special_tokens=True)
        # Use a pipeline as a high-level helper
        translator = pipeline('translation', model=model, tokenizer=tokenizer, src_lang=self.sl, tgt_lang=self.tl)
        translated_text = translator(text, max_length=512)
        return translated_text[0]['translation_text']
    
    def flan(self):
        tokenizer = T5Tokenizer.from_pretrained(self.model_name, legacy=False)
        model = T5ForConditionalGeneration.from_pretrained(self.model_name)
        prompt = f"translate {self.sl} to {self.tl}: {self.input_text}"
        input_ids = tokenizer(prompt, return_tensors="pt").input_ids
        outputs = model.generate(input_ids)
        return tokenizer.decode(outputs[0], skip_special_tokens=True).strip()

    def tfive(self):
        tokenizer = T5Tokenizer.from_pretrained(self.model_name)
        model = T5ForConditionalGeneration.from_pretrained(self.model_name, device_map="auto")
        prompt = f"translate {self.sl} to {self.tl}: {self.input_text}"
        input_ids = tokenizer.encode(prompt, return_tensors="pt")
        output_ids = model.generate(input_ids, max_length=512)
        translated_text = tokenizer.decode(output_ids[0], skip_special_tokens=True).strip()
        return translated_text
    
    def mbart_many_to_many(self):
        from transformers import MBartForConditionalGeneration, MBart50TokenizerFast
        model = MBartForConditionalGeneration.from_pretrained(self.model_name)
        tokenizer = MBart50TokenizerFast.from_pretrained(self.model_name)
        # translate source to target
        tokenizer.src_lang = languagecodes.mbart_large_languages[self.sl]
        encoded = tokenizer(self.input_text, return_tensors="pt")
        generated_tokens = model.generate(
            **encoded,
            forced_bos_token_id=tokenizer.lang_code_to_id[languagecodes.mbart_large_languages[self.tl]]
        )
        return tokenizer.batch_decode(generated_tokens, skip_special_tokens=True)[0]
    
    def mbart_one_to_many(self):
        # translate from English
        from transformers import MBartForConditionalGeneration, MBart50TokenizerFast
        model = MBartForConditionalGeneration.from_pretrained(self.model_name)
        tokenizer = MBart50TokenizerFast.from_pretrained(self.model_name, src_lang="en_XX")
        model_inputs = tokenizer(self.input_text, return_tensors="pt")
        langid = languagecodes.mbart_large_languages[self.tl]
        generated_tokens = model.generate(
            **model_inputs,
            forced_bos_token_id=tokenizer.lang_code_to_id[langid]
        )
        return tokenizer.batch_decode(generated_tokens, skip_special_tokens=True)[0]
    
    def mbart_many_to_one(self):
        # translate to English
        from transformers import MBartForConditionalGeneration, MBart50TokenizerFast
        model = MBartForConditionalGeneration.from_pretrained(self.model_name)
        tokenizer = MBart50TokenizerFast.from_pretrained(self.model_name)
        tokenizer.src_lang = languagecodes.mbart_large_languages[self.sl]
        encoded = tokenizer(self.input_text, return_tensors="pt")
        generated_tokens = model.generate(**encoded)
        return tokenizer.batch_decode(generated_tokens, skip_special_tokens=True)[0]
        
    def mtom(self):
        from transformers import M2M100ForConditionalGeneration, M2M100Tokenizer
        model = M2M100ForConditionalGeneration.from_pretrained(self.model_name)
        tokenizer = M2M100Tokenizer.from_pretrained(self.model_name)
        tokenizer.src_lang = self.sl
        encoded = tokenizer(self.input_text, return_tensors="pt")
        generated_tokens = model.generate(**encoded, forced_bos_token_id=tokenizer.get_lang_id(self.tl))
        return tokenizer.batch_decode(generated_tokens, skip_special_tokens=True)[0]

    def smallonehundred(self):
        from transformers import M2M100ForConditionalGeneration
        from tokenization_small100 import SMALL100Tokenizer
        model = M2M100ForConditionalGeneration.from_pretrained(self.model_name)
        tokenizer = SMALL100Tokenizer.from_pretrained(self.model_name)
        tokenizer.tgt_lang = self.tl
        encoded_sl = tokenizer(self.input_text, return_tensors="pt")
        generated_tokens = model.generate(**encoded_sl, max_length=256, num_beams=5)
        return tokenizer.batch_decode(generated_tokens, skip_special_tokens=True)[0]
    
    def LegoMT(self):
        from transformers import M2M100ForConditionalGeneration, M2M100Tokenizer
        model = M2M100ForConditionalGeneration.from_pretrained(self.model_name) # "Lego-MT/Lego-MT"
        tokenizer = M2M100Tokenizer.from_pretrained(self.model_name)
        tokenizer.src_lang = self.sl
        encoded = tokenizer(self.input_text, return_tensors="pt")
        generated_tokens = model.generate(**encoded, forced_bos_token_id=tokenizer.get_lang_id(self.tl))
        return tokenizer.batch_decode(generated_tokens, skip_special_tokens=True)[0]  
    
    def bigscience(self):  
        tokenizer = AutoTokenizer.from_pretrained(self.model_name)
        model = AutoModelForSeq2SeqLM.from_pretrained(self.model_name)
        self.input_text = self.input_text if self.input_text.endswith('.') else f'{self.input_text}.'
        inputs = tokenizer.encode(f"Translate to {self.tl}: {self.input_text}", return_tensors="pt")
        outputs = model.generate(inputs)
        translation = tokenizer.decode(outputs[0])
        translation = translation.replace('<pad> ', '').replace('</s>', '')
        return translation
    
    def bloomz(self):  
        tokenizer = AutoTokenizer.from_pretrained(self.model_name)
        model = AutoModelForCausalLM.from_pretrained(self.model_name)
        self.input_text = self.input_text if self.input_text.endswith('.') else f'{self.input_text}.'
        # inputs = tokenizer.encode(f"Translate from {self.sl} to {self.tl}: {self.input_text} Translation:", return_tensors="pt")
        inputs = tokenizer.encode(f"Translate to {self.tl}: {self.input_text}", return_tensors="pt")
        outputs = model.generate(inputs)
        translation = tokenizer.decode(outputs[0])
        translation = translation.replace('<pad> ', '').replace('</s>', '')
        translation = translation.split('Translation:')[-1].strip() if 'Translation:' in translation else translation.strip()
        return translation
    
    def nllb(self):
        tokenizer = AutoTokenizer.from_pretrained(self.model_name, src_lang=self.sl)
        # model = AutoModelForSeq2SeqLM.from_pretrained(self.model_name, device_map="auto", torch_dtype=torch.bfloat16)
        model = AutoModelForSeq2SeqLM.from_pretrained(self.model_name)
        translator = pipeline('translation', model=model, tokenizer=tokenizer, src_lang=self.sl, tgt_lang=self.tl)
        translated_text = translator(self.input_text, max_length=512)
        return translated_text[0]['translation_text']

    def seamlessm4t1(self):
        from transformers import AutoProcessor, SeamlessM4TModel
        processor = AutoProcessor.from_pretrained(self.model_name)
        model = SeamlessM4TModel.from_pretrained(self.model_name)
        src_lang = iso1toall.get(self.sl)[2] # 'deu', 'ron', 'eng', 'fra'        
        tgt_lang = iso1toall.get(self.tl)[2]
        text_inputs = processor(text = self.input_text, src_lang=src_lang, return_tensors="pt")
        output_tokens = model.generate(**text_inputs, tgt_lang=tgt_lang, generate_speech=False) 
        return processor.decode(output_tokens[0].tolist()[0], skip_special_tokens=True)
        
    def seamlessm4t2(self):
        from transformers import AutoProcessor, SeamlessM4Tv2ForTextToText
        processor = AutoProcessor.from_pretrained(self.model_name)
        model = SeamlessM4Tv2ForTextToText.from_pretrained(self.model_name)
        src_lang = iso1toall.get(self.sl)[2] # 'deu', 'ron', 'eng', 'fra'        
        tgt_lang = iso1toall.get(self.tl)[2]
        text_inputs = processor(text=self.input_text, src_lang=src_lang, return_tensors="pt")
        decoder_input_ids = model.generate(**text_inputs, tgt_lang=tgt_lang)[0].tolist()
        return processor.decode(decoder_input_ids, skip_special_tokens=True)
    
    def wingpt(self):
        model = AutoModelForCausalLM.from_pretrained(
           self.model_name,
           torch_dtype="auto",
           device_map="auto"
        )
        tokenizer = AutoTokenizer.from_pretrained(self.model_name)
        # input_json = '{"input_text": self.input_text}'
        messages = [
           {"role": "system", "content": f"Translate this to {self.tl} language"}, 
           {"role": "user", "content": self.input_text}
        ]
        
        text = tokenizer.apply_chat_template(
           messages,
           tokenize=False,
           add_generation_prompt=True
        )
        model_inputs = tokenizer([text], return_tensors="pt").to(model.device)
        
        generated_ids = model.generate(
           **model_inputs,
           max_new_tokens=512,
           temperature=0.1
        )
        
        generated_ids = [
           output_ids[len(input_ids):] for input_ids, output_ids in zip(model_inputs.input_ids, generated_ids)
        ]
        output = tokenizer.batch_decode(generated_ids, skip_special_tokens=True)[0]
        result = output.split('\n')[-1].strip() if '\n' in output else output.strip()
        return result
    
    def eurollm(self):
        tokenizer = AutoTokenizer.from_pretrained(self.model_name)
        model = AutoModelForCausalLM.from_pretrained(self.model_name)  
        prompt = f"{self.sl}: {self.input_text} {self.tl}:"
        inputs = tokenizer(prompt, return_tensors="pt")
        outputs = model.generate(**inputs, max_new_tokens=512)
        output = tokenizer.decode(outputs[0], skip_special_tokens=True)
        print(output)
        # result = output.rsplit(f'{self.tl}:')[-1].strip() if f'{self.tl}:' in output else output.strip()
        result = output.rsplit(f'{self.tl}:')[-1].strip() if '\n' in output or f'{self.tl}:' in output else output.strip()
        return result
    
    def eurollm_instruct(self):
        tokenizer = AutoTokenizer.from_pretrained(self.model_name)
        model = AutoModelForCausalLM.from_pretrained(self.model_name)
        text = f'<|im_start|>system\n<|im_end|>\n<|im_start|>user\nTranslate the following {self.sl} source text to {self.tl}:\n{self.sl}: {self.input_text} \n{self.tl}: <|im_end|>\n<|im_start|>assistant\n'
        inputs = tokenizer(text, return_tensors="pt")
        outputs = model.generate(**inputs, max_new_tokens=512)
        output = tokenizer.decode(outputs[0], skip_special_tokens=True)
        if f'{self.tl}:' in output:
            output = output.rsplit(f'{self.tl}:')[-1].strip().replace('assistant\n', '').strip()
        return output

    def unbabel(self):
        pipe = pipeline("text-generation", model=self.model_name, torch_dtype=torch.bfloat16, device_map="auto")
        messages = [{"role": "user",
                     "content": f"Translate the following text from {self.sl} into {self.tl}.\n{self.sl}: {self.input_text}.\n{self.tl}:"}]
        prompt = pipe.tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=False)
        tokenized_input = pipe.tokenizer(self.input_text, return_tensors="pt")
        num_input_tokens = len(tokenized_input["input_ids"][0])
        max_new_tokens = round(num_input_tokens + 0.5 * num_input_tokens)
        outputs = pipe(prompt, max_new_tokens=max_new_tokens, do_sample=False)
        translated_text = outputs[0]["generated_text"]
        print(f"Input chars: {len(self.input_text)}", f"Input tokens: {num_input_tokens}", f"max_new_tokens: {max_new_tokens}",
              "Chars to tokens ratio:", round(len(self.input_text) / num_input_tokens, 2), f"Raw translation: {translated_text}")
        markers = ["<end_of_turn>", "<|im_end|>", "<|im_start|>assistant"] # , "\n" 
        for marker in markers:
            if marker in translated_text:
                translated_text = translated_text.split(marker)[1].strip()
        translated_text = translated_text.replace('Answer:', '', 1).strip() if translated_text.startswith('Answer:') else translated_text
        translated_text = translated_text.split("Translated text:")[0].strip() if "Translated text:" in translated_text else translated_text
        split_translated_text = translated_text.split('\n', translated_text.count('\n'))
        translated_text = '\n'.join(split_translated_text[:self.input_text.count('\n')+1])
        return translated_text

    def bergamot(model_name: str = 'deen', sl: str = 'de', tl: str = 'en', input_text: str = 'Hallo, mein Freund'):
        try:
            import bergamot
            # input_text = [input_text] if isinstance(input_text, str) else input_text           
            config = bergamot.ServiceConfig(numWorkers=4)
            service = bergamot.Service(config)
            model = service.modelFromConfigPath(f"./{model_name}/bergamot.config.yml")
            options = bergamot.ResponseOptions(alignment=False, qualityScores=False, HTML=False)
            rawresponse = service.translate(model, bergamot.VectorString(input_text), options)
            translated_text: str = next(iter(rawresponse)).target.text
            message_text = f"Translated from {sl} to {tl} with Bergamot {model_name}."
        except Exception as error:
            response = error
        return translated_text, message_text

@timer
@spaces.GPU
def translate_text(input_text: str, s_language: str, t_language: str, model_name: str) -> tuple[str, str]:
    """
    Translates the input text from the source language to the target language  using a specified model.

    Parameters:
        input_text (str): The source text to be translated
        s_language (str): The source language of the input text
        t_language (str): The target language in which the input text is translated
        model_name (str): The selected translation model name

    Returns:
        tuple: 
            translated_text(str): The input text translated to the selected target language
            message_text(str):  A descriptive message summarizing the translation process. Example: "Translated from English to German with Helsinki-NLP."
    
    Example:
        >>> translate_text("Hello world", "English", "German", "Helsinki-NLP")
        ("Hallo Welt", "Translated from English to German with Helsinki-NLP.")
    """
    
    sl = all_langs[s_language][0]
    tl = all_langs[t_language][0]
    message_text = f'Translated from {s_language} to {t_language} with {model_name}'
    if not input_text or input_text.strip() == '':
        translated_text = f'No input text entered!'
        message_text = 'Please enter a text to translate!'
        return translated_text, message_text
    if sl == tl:
        translated_text = f'Source language {s_language} identical to target language {t_language}!'
        message_text = 'Please choose different target and source language!'
        return translated_text, message_text 
    try:    
        if "-mul" in model_name.lower() or "mul-" in model_name.lower() or "-roa" in model_name.lower():
            translated_text, message_text = Translators(model_name, sl, tl, input_text).HelsinkiNLP_mulroa()
        
        elif model_name == "Helsinki-NLP":
            translated_text, message_text = Translators(model_name, sl, tl, input_text).HelsinkiNLP()
        
        elif model_name == 'Argos':
            translated_text = Translators(model_name, sl, tl, input_text).argos()

        elif model_name == "QUICKMT":
            translated_text, message_text = Translators(model_name, sl, tl, input_text).quickmt()
    
        elif model_name == 'Google':
            translated_text = Translators(model_name, sl, tl, input_text).google()
    
        elif model_name == "Helsinki-NLP/opus-mt-tc-bible-big-roa-en":
            translated_text, message_text = Translators(model_name, sl, tl, input_text).simplepipe()
        
        elif 'mitre' in model_name.lower():
            translated_text = Translators(model_name, sl, tl, input_text).mitre()
        
        elif "m2m" in model_name.lower():
            translated_text = Translators(model_name, sl, tl, input_text).mtom()
        
        elif "small100" in model_name.lower():
            translated_text = Translators(model_name, sl, tl, input_text).smallonehundred()

        elif "rosetta" in model_name.lower():
            translated_text = Translators(model_name, s_language, t_language, input_text).rosetta()
        
        elif "lego" in model_name.lower():
            translated_text = Translators(model_name, sl, tl, input_text).LegoMT()

        elif "niutrans" in model_name.lower():
            translated_text = Translators(model_name, sl, tl, input_text).niutrans()

        elif "salamandra" in model_name.lower():
            translated_text = Translators(model_name, s_language, t_language, input_text).salamandratapipe()
        
        elif model_name.startswith('google-t5'):
            translated_text = Translators(model_name, s_language, t_language, input_text).tfive()
            
        elif 'flan' in model_name.lower():
            translated_text = Translators(model_name, s_language, t_language, input_text).flan()

        elif 'madlad' in model_name.lower():
            translated_text = Translators(model_name, sl, tl, input_text).madlad()
            
        elif 'mt0' in model_name.lower():
            translated_text = Translators(model_name, s_language, t_language, input_text).bigscience()
    
        elif 'bloomz' in model_name.lower():
            translated_text = Translators(model_name, s_language, t_language, input_text).bloomz()
            
        elif 'nllb' in model_name.lower():
            nnlbsl, nnlbtl = languagecodes.nllb_language_codes[s_language], languagecodes.nllb_language_codes[t_language]
            translated_text = Translators(model_name, nnlbsl, nnlbtl, input_text).nllb()
        
        elif model_name == "facebook/mbart-large-50-many-to-many-mmt":
            translated_text = Translators(model_name, s_language, t_language, input_text).mbart_many_to_many()
    
        elif model_name == "facebook/mbart-large-50-one-to-many-mmt":
            translated_text = Translators(model_name, s_language, t_language, input_text).mbart_one_to_many()
    
        elif model_name == "facebook/mbart-large-50-many-to-one-mmt":
            translated_text = Translators(model_name, s_language, t_language, input_text).mbart_many_to_one()
          
        elif model_name ==  "facebook/seamless-m4t-v2-large":
            translated_text = Translators(model_name, sl, tl, input_text).seamlessm4t2()
          
        elif "m4t-medium" in model_name or "m4t-large" in model_name:
            translated_text = Translators(model_name, sl, tl, input_text).seamlessm4t1()
        
        elif model_name == "utter-project/EuroLLM-1.7B-Instruct":
            translated_text = Translators(model_name, s_language, t_language, input_text).eurollm_instruct()
        
        elif model_name == "utter-project/EuroLLM-1.7B":
            translated_text = Translators(model_name, s_language, t_language, input_text).eurollm()
                
        elif 'Unbabel' in model_name:   
            translated_text = Translators(model_name, s_language, t_language, input_text).unbabel()
            
        elif "winninghealth/WiNGPT" in model_name:      
            translated_text = Translators(model_name, s_language, t_language, input_text).wingpt()

        elif "HPLT" in model_name:
            if model_name == "HPLT-OPUS":
                translated_text, message_text = Translators(model_name, sl, tl, input_text).hplt(opus = True)
            else:
                translated_text, message_text = Translators(model_name, sl, tl, input_text).hplt()
        
        elif model_name == "Bergamot":
            translated_text, message_text = Translators(model_name, s_language, t_language, input_text).bergamot()
            
    except Exception as trerror:
        translated_text = f'Error in main function "translate_text": {trerror}'
    finally:
        print(input_text, translated_text, message_text)
        return translated_text, message_text

def swap_languages(src_lang, tgt_lang):
    '''Swap dropdown values for source and target language'''
    return tgt_lang, src_lang 

def get_info(model_name: str, sl: str = None, tl: str = None):
    helsinki = '### [Helsinki-NLP](https://huggingface.co/Helsinki-NLP "Helsinki-NLP")'
    if model_name == "Helsinki-NLP" and sl and tl:
        url = f'https://huggingface.co/{model_name}/opus-mt-{sl}-{tl}/raw/main/README.md'
        response = httpx.get(url).text
        if 'Repository not found' in response or 'Invalid username or password' in response:
            return helsinki
        return response
    elif model_name == "Argos":
        return httpx.get(f'https://huggingface.co/TiberiuCristianLeon/Argostranslate/raw/main/README.md').text
    elif "HPLT" in model_name:
        return """[HPLT Uni direction translation models](https://huggingface.co/collections/HPLT/hplt-12-uni-direction-translation-models)
            ['ar-en', 'bs-en', 'ca-en', 'en-ar', 'en-bs', 'en-ca', 'en-et', 'en-eu', 'en-fi',
            'en-ga', 'en-gl', 'en-hi', 'en-hr', 'en-is', 'en-mt', 'en-nn', 'en-sq', 'en-sw',
            'en-zh_hant', 'et-en', 'eu-en', 'fi-en', 'ga-en', 'gl-en', 'hi-en', 'hr-en',
            'is-en', 'mt-en', 'nn-en', 'sq-en', 'sw-en', 'zh_hant-en']"""
    elif "QUICKMT" in model_name:
        return """[QUICKMT](https://huggingface.co/quickmt)  
            ['ar', 'bn', 'cs', 'da', 'de', 'el', 'en', 'es', 'fa', 'fr', 'he',
           'hi', 'hu', 'id', 'it', 'is', 'ja', 'ko', 'lv', 'pl', 'pt', 'ro', 'ru',
           'sv', 'th', 'tr', 'ur', 'vi', 'zh']"""
    elif model_name == "Google":
        return "Google Translate Online"
    else:
        return httpx.get(f'https://huggingface.co/{model_name}/raw/main/README.md').text

with gr.Blocks() as interface:
    gr.Markdown("### Machine Text Translation with Gradio API and MCP Server")
    input_text = gr.Textbox(label="Enter text to translate:", placeholder="Type your text here, maximum 512 tokens",
                            autofocus=True, submit_btn='Translate', max_length=512) 
    with gr.Row(variant="compact"):
        s_language = gr.Dropdown(choices=langs, value = DEFAULTS[0], label="Source language", interactive=True, scale=2)
        t_language = gr.Dropdown(choices=langs, value = DEFAULTS[1], label="Target language", interactive=True, scale=2)
        swap_btn = gr.Button("Swap Languages", size="md", scale=1)
        swap_btn.click(fn=swap_languages, inputs=[s_language, t_language], outputs=[s_language, t_language], api_visibility="private")
    # with gr.Row(equal_height=True):
        model_name = gr.Dropdown(choices=models, label=f"Select a model. Default is {DEFAULTS[2]}.", value=DEFAULTS[2], interactive=True, scale=2)
        # translate_btn = gr.Button(value="Translate", scale=1)

    translated_text = gr.Textbox(label="Translated text:", placeholder="Display field for translation", interactive=False, buttons=["copy"], lines=2)
    message_text = gr.Textbox(label="Messages:", placeholder="Display field for status and error messages", interactive=False,
                          value=f'Default translation settings: from {s_language.value} to {t_language.value} with {model_name.value}.', lines=2)
    allmodels = gr.HTML(label="Models with links:", value=', '.join([f'<a href="https://huggingface.co/{model}">{model}</a>' for model in models]),
                        show_label=False, container=True, css_template="""a {padding: 0px;}""")
    model_info = gr.Markdown(label="Model info:", value=get_info(DEFAULTS[2], DEFAULTS[0], DEFAULTS[1]), buttons=["copy"])

    model_name.change(fn=get_info, inputs=[model_name, s_language, t_language], outputs=model_info, api_visibility="private")
    # translate_btn.click(
    #     fn=translate_text, 
    #     inputs=[input_text, s_language, t_language, model_name], 
    #     outputs=[translated_text, message_text]
    # )
    input_text.submit(
    fn=translate_text,
    inputs=[input_text, s_language, t_language, model_name],
    outputs=[translated_text, message_text]
    )
if __name__ == "__main__":
    interface.launch(mcp_server=True, footer_links=["api", "settings"])
    # interface.queue().launch(server_name="0.0.0.0", show_error=True, server_port=7860, mcp_server=True)