[Update] translation: Add Plamo API support with authentication and translation functionality

src-python/config.py

@@ -1184,6 +1184,7 @@ class Config:
         self._OSC_PORT = 9000
         self._AUTH_KEYS = {
             "DeepL_API": None,
+            "Plamo_API": None,
         }
         self._USE_EXCLUDE_WORDS = True
         self._SELECTED_TRANSLATION_COMPUTE_DEVICE = copy.deepcopy(self.SELECTABLE_COMPUTE_DEVICE_LIST[0])

src-python/controller.py

@@ -1431,6 +1431,50 @@ class Controller:
         self.updateTranslationEngineAndEngineList()
         return {"status":200, "result":config.AUTH_KEYS[translator_name]}
 
+    def getPlamoAuthKey(self, *args, **kwargs) -> dict:
+        return {"status":200, "result":config.AUTH_KEYS["Plamo_API"]}
+
+    def setPlamoAuthKey(self, data, *args, **kwargs) -> dict:
+        printLog("Set Plamo Auth Key", data)
+        translator_name = "Plamo_API"
+        try:
+            data = str(data)
+            if len(data) == 32:
+                key = data
+                auth_keys = config.AUTH_KEYS
+                auth_keys[translator_name] = key
+                config.AUTH_KEYS = auth_keys
+                config.SELECTABLE_TRANSLATION_ENGINE_STATUS[translator_name] = True
+                self.updateTranslationEngineAndEngineList()
+                response = {"status":200, "result":config.AUTH_KEYS[translator_name]}
+            else:
+                response = {
+                    "status":400,
+                    "result":{
+                        "message":"Plamo auth key length is not correct",
+                        "data": config.AUTH_KEYS[translator_name]
+                    }
+                }
+        except Exception as e:
+            errorLogging()
+            response = {
+                "status":400,
+                "result":{
+                    "message":f"Error {e}",
+                    "data": config.AUTH_KEYS[translator_name]
+                }
+            }
+        return response
+
+    def delPlamoAuthKey(self, *args, **kwargs) -> dict:
+        translator_name = "Plamo_API"
+        auth_keys = config.AUTH_KEYS
+        auth_keys[translator_name] = None
+        config.AUTH_KEYS = auth_keys
+        config.SELECTABLE_TRANSLATION_ENGINE_STATUS[translator_name] = False
+        self.updateTranslationEngineAndEngineList()
+        return {"status":200, "result":config.AUTH_KEYS[translator_name]}
+
     @staticmethod
     def getCtranslate2WeightType(*args, **kwargs) -> dict:
         return {"status":200, "result":config.CTRANSLATE2_WEIGHT_TYPE}
@@ -2257,6 +2301,17 @@ class Controller:
                             auth_keys = config.AUTH_KEYS
                             auth_keys[engine] = None
                             config.AUTH_KEYS = auth_keys
+                case "Plamo_API":
+                    printLog("Start check Plamo API Key")
+                    config.SELECTABLE_TRANSLATION_ENGINE_STATUS[engine] = False
+                    if config.AUTH_KEYS[engine] is not None:
+                        if model.authenticationTranslatorPlamoAuthKey(auth_key=config.AUTH_KEYS[engine]) is True:
+                            config.SELECTABLE_TRANSLATION_ENGINE_STATUS[engine] = True
+                        else:
+                            # error update Auth key
+                            auth_keys = config.AUTH_KEYS
+                            auth_keys[engine] = None
+                            config.AUTH_KEYS = auth_keys
                 case _:
                     if connected_network is True:
                         config.SELECTABLE_TRANSLATION_ENGINE_STATUS[engine] = True

src-python/mainloop.py

@@ -168,6 +168,10 @@ mapping = {
     "/set/data/deepl_auth_key": {"status": False, "variable":controller.setDeeplAuthKey},
     "/delete/data/deepl_auth_key": {"status": False, "variable":controller.delDeeplAuthKey},
 
+    "/get/data/plamo_auth_key": {"status": False, "variable":controller.getPlamoAuthKey},
+    "/set/data/plamo_auth_key": {"status": False, "variable":controller.setPlamoAuthKey},
+    "/delete/data/plamo_auth_key": {"status": False, "variable":controller.delPlamoAuthKey},
+
     "/get/data/convert_message_to_romaji": {"status": True, "variable":controller.getConvertMessageToRomaji},
     "/set/enable/convert_message_to_romaji": {"status": True, "variable":controller.setEnableConvertMessageToRomaji},
     "/set/disable/convert_message_to_romaji": {"status": True, "variable":controller.setDisableConvertMessageToRomaji},

src-python/model.py

@@ -140,6 +140,10 @@ class Model:
         result = self.translator.authenticationDeepLAuthKey(auth_key)
         return result
 
+    def authenticationTranslatorPlamoAuthKey(self, auth_key):
+        result = self.translator.authenticationPlamoAuthKey(auth_key)
+        return result
+
     def startLogger(self):
         os_makedirs(config.PATH_LOGS, exist_ok=True)
         file_name = os_path.join(config.PATH_LOGS, f"{datetime.now().strftime('%Y-%m-%d_%H-%M-%S')}.log")

src-python/models/translation/translation_languages.py

@@ -603,4 +603,42 @@ translation_lang["nllb-200-distilled-1.3B-ct2-int8"] = {
 translation_lang["nllb-200-3.3B-ct2-int8"] = {
     "source":dict_nllb_languages,
     "target":dict_nllb_languages,
+}
+
+dict_plamo_languages = {
+    "English": "English",
+    "Japanese": "Japanese",
+    "Korean": "Korean",
+    "French": "French",
+    "German": "German",
+    "Spanish": "Spanish",
+    "Portuguese": "Portuguese",
+    "Russian": "Russian",
+    "Italian": "Italian",
+    "Dutch": "Dutch",
+    "Polish": "Polish",
+    "Turkish": "Turkish",
+    "Arabic": "Arabic",
+    "Hindi": "Hindi",
+    "Thai": "Thai",
+    "Vietnamese": "Vietnamese",
+    "Indonesian": "Indonesian",
+    "Malay": "Malay",
+    "Filipino": "Filipino",
+    "Swedish": "Swedish",
+    "Finnish": "Finnish",
+    "Danish": "Danish",
+    "Norwegian": "Norwegian",
+    "Romanian": "Romanian",
+    "Czech": "Czech",
+    "Hungarian": "Hungarian",
+    "Greek": "Greek",
+    "Hebrew": "Hebrew",
+    "Simplified Chinese":"Simplified Chinese",
+    "Traditional Chinese":"Traditional Chinese"
+}
+
+translation_lang["Plamo_API"] = {
+    "source":dict_plamo_languages,
+    "target":dict_plamo_languages,
 }

src-python/models/translation/translation_plamo.py

@@ -0,0 +1,80 @@
+from langchain_openai import ChatOpenAI
+
+class PlamoClient:
+    def __init__(self, api_key: str):
+        self.api_key = api_key
+        self.base_url = "https://api.platform.preferredai.jp/v1"
+        self.model = "plamo-2.0-prime"
+        self.supported_languages = """
+        English
+        Japanese
+        Korean
+        French
+        German
+        Spanish
+        Portuguese
+        Russian
+        Italian
+        Dutch
+        Polish
+        Turkish
+        Arabic
+        Hindi
+        Thai
+        Vietnamese
+        Indonesian
+        Malay
+        Filipino
+        Swedish
+        Finnish
+        Danish
+        Norwegian
+        Romanian
+        Czech
+        Hungarian
+        Greek
+        Hebrew
+        Simplified Chinese
+        Traditional Chinese
+        """
+        self.prompt_template = f"""
+        You are a translation assistant that uses the `plamo-translate` tool.
+        Translate the following text.Supported languages include:{self.supported_languages}
+        Translate the following text from {{input_lang}} to {{output_lang}}.
+        output only the translated text without any additional commentary.
+        """
+        self.plamo_llm = ChatOpenAI(
+            base_url=self.base_url,
+            model=self.model,
+            streaming=True,
+            openai_api_key=self.api_key,
+        )
+
+    def translate_text(self, text: str, input_lang: str, output_lang: str):
+        messages = [
+            {
+                "role": "system",
+                "content": self.prompt_template.format(
+                    input_lang=input_lang, output_lang=output_lang
+                ),
+            },
+            {"role": "user", "content": text},
+        ]
+
+        output = ""
+        for chunk in self.plamo_llm.stream(messages):
+            output += chunk.content
+
+        return output[:-1]
+
+
+if __name__ == "__main__":
+    text = """
+        毎朝コーヒーを入れるのがささやかな楽しみになってる
+        """
+    input_lang = "Japanese"
+    output_lang = "English"
+
+    plamo_client = PlamoClient(api_key="AUTH_KEY")
+    translated_text = plamo_client.translate_text(text, input_lang, output_lang)
+    print(translated_text)

src-python/models/translation/translation_translator.py

@@ -9,12 +9,14 @@ except Exception:
 try:
     from .translation_languages import translation_lang
     from .translation_utils import ctranslate2_weights
+    from .translation_plamo import PlamoClient
 except Exception:
     import sys
     print(os_path.dirname(os_path.dirname(os_path.dirname(os_path.abspath(__file__)))))
     sys.path.append(os_path.dirname(os_path.dirname(os_path.dirname(os_path.abspath(__file__)))))
     from translation_languages import translation_lang
     from translation_utils import ctranslate2_weights
+    from translation_plamo import PlamoClient
 
 import ctranslate2
 import transformers
@@ -27,6 +29,7 @@ warnings.filterwarnings("ignore")
 class Translator():
     def __init__(self):
         self.deepl_client = None
+        self.plamo_client = None
         self.ctranslate2_translator = None
         self.ctranslate2_tokenizer = None
         self.is_loaded_ctranslate2_model = False
@@ -36,13 +39,24 @@ class Translator():
         result = True
         try:
             self.deepl_client = DeepLClient(authkey)
-            self.deepl_client.translate_text(" ", target_lang="EN-US")
+            self.deepl_client.translate_text("Hello World", target_lang="EN-US")
         except Exception:
             errorLogging()
             self.deepl_client = None
             result = False
         return result
 
+    def authenticationPlamoAuthKey(self, authkey):
+        result = True
+        try:
+            self.plamo_client = PlamoClient(authkey)
+            self.plamo_client.translate_text("Hello World", target_lang="English")
+        except Exception:
+            errorLogging()
+            self.plamo_client = None
+            result = False
+        return result
+
     def changeCTranslate2Model(self, path, model_type, device="cpu", device_index=0):
         self.is_loaded_ctranslate2_model = False
         directory_name = ctranslate2_weights[model_type]["directory_name"]
@@ -135,6 +149,15 @@ class Translator():
                                 source_lang=source_language,
                                 target_lang=target_language,
                                 ).text
+                case "Plamo_API":
+                    if self.plamo_client is None:
+                        result = False
+                    else:
+                        result = self.plamo_client.translate_text(
+                            message,
+                            input_lang=source_language,
+                            output_lang=target_language,
+                            )
                 case "Google":
                     if self.is_enable_translators is True:
                         result = other_web_Translator(
@@ -170,7 +193,7 @@ class Translator():
             errorLogging()
             result = False
         return result
-    
+
 if __name__ == "__main__":
     translator = Translator()
     # test CTranslate2 model nllb-200-distilled-1.3B-ct2-int8