every_holiday/VRCT
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

🐛[bugfix] controller: offline時の処理を追加

Browse Source
This commit is contained in:
misyaguziya
2025-02-06 18:29:07 +09:00
parent 69d9e9bad3
commit 78d79f4a54
6 changed files with 155 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
src-python/config.py
View File

@@ -237,6 +237,15 @@ class Config:
if isinstance(value, dict):
self._SELECTABLE_TRANSLATION_ENGINE_STATUS = value
@property
def SELECTABLE_TRANSCRIPTION_ENGINE_STATUS(self):
return self._SELECTABLE_TRANSCRIPTION_ENGINE_STATUS
@SELECTABLE_TRANSCRIPTION_ENGINE_STATUS.setter
def SELECTABLE_TRANSCRIPTION_ENGINE_STATUS(self, value):
if isinstance(value, dict):
self._SELECTABLE_TRANSCRIPTION_ENGINE_STATUS = value
# Save Json Data
## Main Window
@property
@@ -977,6 +986,9 @@ class Config:
self._SELECTABLE_TRANSLATION_ENGINE_STATUS = {}
for engine in self.SELECTABLE_TRANSLATION_ENGINE_LIST:
self._SELECTABLE_TRANSLATION_ENGINE_STATUS[engine] = False
self._SELECTABLE_TRANSCRIPTION_ENGINE_STATUS = {}
for engine in self.SELECTABLE_TRANSCRIPTION_ENGINE_LIST:
self._SELECTABLE_TRANSCRIPTION_ENGINE_STATUS[engine] = False
# Save Json Data
## Main Window

104
src-python/controller.py
View File

@@ -6,7 +6,7 @@ import re
from device_manager import device_manager
from config import config
from model import model
from utils import removeLog, printLog, errorLogging
from utils import removeLog, printLog, errorLogging, isConnectedNetwork
class Controller:
def __init__(self) -> None:
@@ -25,6 +25,18 @@ class Controller:
self.run = run
# response functions
def connectedNetwork(self) -> None:
self.run(
200,
self.run_mapping["connected_network"],
)
def disconnectedNetwork(self) -> None:
self.run(
200,
self.run_mapping["disconnected_network"],
)
def updateMicHostList(self) -> None:
self.run(
200,
@@ -148,15 +160,25 @@ class Controller:
)
def downloaded(self) -> None:
weight_type_dict = config.SELECTABLE_CTRANSLATE2_WEIGHT_TYPE_DICT
weight_type_dict[self.weight_type] = True
config.SELECTABLE_CTRANSLATE2_WEIGHT_TYPE_DICT = weight_type_dict
if model.checkTranslatorCTranslate2ModelWeight(self.weight_type) is True:
weight_type_dict = config.SELECTABLE_CTRANSLATE2_WEIGHT_TYPE_DICT
weight_type_dict[self.weight_type] = True
config.SELECTABLE_CTRANSLATE2_WEIGHT_TYPE_DICT = weight_type_dict
self.run(
200,
self.run_mapping["downloaded_ctranslate2_weight"],
self.weight_type,
)
self.run(
200,
self.run_mapping["downloaded_ctranslate2_weight"],
self.weight_type,
)
else:
self.run(
400,
self.run_mapping["error_ctranslate2_weight"],
{
"message":"CTranslate2 weight download error",
"data": None
},
)
class DownloadWhisper:
def __init__(self, run_mapping:dict, weight_type:str, run:Callable[[int, str, Any], None]) -> None:
@@ -173,15 +195,25 @@ class Controller:
)
def downloaded(self) -> None:
weight_type_dict = config.SELECTABLE_WHISPER_WEIGHT_TYPE_DICT
weight_type_dict[self.weight_type] = True
config.SELECTABLE_WHISPER_WEIGHT_TYPE_DICT = weight_type_dict
if model.checkTranscriptionWhisperModelWeight(self.weight_type) is True:
weight_type_dict = config.SELECTABLE_WHISPER_WEIGHT_TYPE_DICT
weight_type_dict[self.weight_type] = True
config.SELECTABLE_WHISPER_WEIGHT_TYPE_DICT = weight_type_dict
self.run(
200,
self.run_mapping["downloaded_whisper_weight"],
self.weight_type,
)
self.run(
200,
self.run_mapping["downloaded_whisper_weight"],
self.weight_type,
)
else:
self.run(
400,
self.run_mapping["error_whisper_weight"],
{
"message":"Whisper weight download error",
"data": None
},
)
def micMessage(self, result: dict) -> None:
message = result["text"]
@@ -541,6 +573,11 @@ class Controller:
self.updateTranslationEngineAndEngineList()
return {"status":200, "result":config.SELECTED_TARGET_LANGUAGES}
@staticmethod
def getTranscriptionEngines(*args, **kwargs) -> dict:
engines = [key for key, value in config.SELECTABLE_TRANSCRIPTION_ENGINE_STATUS.items() if value is True]
return {"status":200, "result":engines}
@staticmethod
def getSelectedTranscriptionEngine(*args, **kwargs) -> dict:
return {"status":200, "result":config.SELECTED_TRANSCRIPTION_ENGINE}
@@ -1665,6 +1702,11 @@ class Controller:
def init(self, *args, **kwargs) -> None:
removeLog()
printLog("Start Initialization")
connected_network = isConnectedNetwork()
if connected_network is True:
self.connectedNetwork()
else:
self.disconnectedNetwork()
printLog("Init Translation Engine Status")
for engine in config.SELECTABLE_TRANSLATION_ENGINE_LIST:
@@ -1680,9 +1722,6 @@ class Controller:
auth_keys = config.AUTH_KEYS
auth_keys[engine] = None
config.AUTH_KEYS = auth_keys
case _:
config.SELECTABLE_TRANSLATION_ENGINE_STATUS[engine] = True
self.initializationProgress(1)
# download CTranslate2 Model Weight
@@ -1708,6 +1747,31 @@ class Controller:
if isinstance(th_download_whisper, Thread):
th_download_whisper.join()
for engine in config.SELECTABLE_TRANSLATION_ENGINE_LIST:
match engine:
case "CTranslate2":
if model.checkTranslatorCTranslate2ModelWeight(config.CTRANSLATE2_WEIGHT_TYPE) is True:
config.SELECTABLE_TRANSLATION_ENGINE_STATUS[engine] = True
else:
config.SELECTABLE_TRANSLATION_ENGINE_STATUS[engine] = False
case _:
if connected_network is True:
config.SELECTABLE_TRANSLATION_ENGINE_STATUS[engine] = True
else:
config.SELECTABLE_TRANSLATION_ENGINE_STATUS[engine] = False
for engine in config.SELECTABLE_TRANSCRIPTION_ENGINE_LIST:
match engine:
case "Whisper":
if model.checkTranscriptionWhisperModelWeight(config.WHISPER_WEIGHT_TYPE) is True:
config.SELECTABLE_TRANSCRIPTION_ENGINE_STATUS[engine] = True
else:
config.SELECTABLE_TRANSCRIPTION_ENGINE_STATUS[engine] = False
case _:
if connected_network is True:
config.SELECTABLE_TRANSCRIPTION_ENGINE_STATUS[engine] = True
else:
config.SELECTABLE_TRANSCRIPTION_ENGINE_STATUS[engine] = False
self.initializationProgress(2)
# set Translation Engine

6
src-python/mainloop.py
View File

@@ -8,6 +8,9 @@ from controller import Controller
from utils import printLog, printResponse, errorLogging, encodeBase64
run_mapping = {
"connected_network":"/run/connected_network",
"disconnected_network":"/run/disconnected_network",
"transcription_mic":"/run/transcription_send_mic_message",
"transcription_speaker":"/run/transcription_receive_speaker_message",
@@ -20,8 +23,10 @@ run_mapping = {
"download_progress_ctranslate2_weight":"/run/download_progress_ctranslate2_weight",
"downloaded_ctranslate2_weight":"/run/downloaded_ctranslate2_weight",
"error_ctranslate2_weight":"/run/error_ctranslate2_weight",
"download_progress_whisper_weight":"/run/download_progress_whisper_weight",
"downloaded_whisper_weight":"/run/downloaded_whisper_weight",
"error_whisper_weight":"/run/error_whisper_weight",
"selected_mic_device":"/run/selected_mic_device",
"selected_speaker_device":"/run/selected_speaker_device",
@@ -79,6 +84,7 @@ mapping = {
"/get/data/selected_target_languages": {"status": True, "variable":controller.getSelectedTargetLanguages},
"/set/data/selected_target_languages": {"status": True, "variable":controller.setSelectedTargetLanguages},
"/get/data/transcription_engines": {"status": False, "variable":controller.getTranscriptionEngines},
"/get/data/selected_transcription_engine": {"status": False, "variable":controller.getSelectedTranscriptionEngine},
"/set/data/selected_transcription_engine": {"status": False, "variable":controller.setSelectedTranscriptionEngine},

2
src-python/model.py
View File

@@ -80,7 +80,7 @@ class Model:
self.speaker_transcriber = None
self.speaker_energy_recorder = None
self.speaker_energy_plot_progressbar = None
self.previous_send_message = ""
self.previous_receive_message = ""
self.translator = Translator()

77
src-python/models/translation/translation_translator.py
View File

@@ -1,6 +1,11 @@
from os import path as os_path
from deepl import Translator as deepl_Translator
from translators import translate_text as other_web_Translator
try:
from translators import translate_text as other_web_Translator
ENABLE_TRANSLATORS = True
except Exception:
ENABLE_TRANSLATORS = False
from .translation_languages import translation_lang
from .translation_utils import ctranslate2_weights
@@ -18,6 +23,7 @@ class Translator():
self.ctranslate2_translator = None
self.ctranslate2_tokenizer = None
self.is_loaded_ctranslate2_model = False
self.is_enable_translators = ENABLE_TRANSLATORS
def authenticationDeepLAuthKey(self, authkey):
result = True
@@ -97,42 +103,47 @@ class Translator():
source_language, target_language = self.getLanguageCode(translator_name, target_country, source_language, target_language)
match translator_name:
case "DeepL":
result = other_web_Translator(
query_text=message,
translator="deepl",
from_language=source_language,
to_language=target_language,
)
if self.is_enable_translators is True:
result = other_web_Translator(
query_text=message,
translator="deepl",
from_language=source_language,
to_language=target_language,
)
case "DeepL_API":
if self.deepl_client is None:
result = False
else:
result = self.deepl_client.translate_text(
message,
source_lang=source_language,
target_lang=target_language,
).text
if self.is_enable_translators is True:
if self.deepl_client is None:
result = False
else:
result = self.deepl_client.translate_text(
message,
source_lang=source_language,
target_lang=target_language,
).text
case "Google":
result = other_web_Translator(
query_text=message,
translator="google",
from_language=source_language,
to_language=target_language,
)
if self.is_enable_translators is True:
result = other_web_Translator(
query_text=message,
translator="google",
from_language=source_language,
to_language=target_language,
)
case "Bing":
result = other_web_Translator(
query_text=message,
translator="bing",
from_language=source_language,
to_language=target_language,
)
if self.is_enable_translators is True:
result = other_web_Translator(
query_text=message,
translator="bing",
from_language=source_language,
to_language=target_language,
)
case "Papago":
result = other_web_Translator(
query_text=message,
translator="papago",
from_language=source_language,
to_language=target_language,
)
if self.is_enable_translators is True:
result = other_web_Translator(
query_text=message,
translator="papago",
from_language=source_language,
to_language=target_language,
)
case "CTranslate2":
result = self.translateCTranslate2(
message=message,

8
src-python/utils.py
View File

@@ -6,6 +6,14 @@ import logging
from logging.handlers import RotatingFileHandler
from ctranslate2 import get_supported_compute_types
import requests
def isConnectedNetwork(url="http://www.google.com", timeout=3):
try:
response = requests.get(url, timeout=timeout)
return response.status_code == 200
except requests.RequestException:
return False
def getBestComputeType(device, device_index) -> str:
compute_types = get_supported_compute_types(device, device_index)