[Update] Refactor message formatting and add validation for message structure

This commit is contained in:
misyaguziya
2025-08-06 20:33:36 +09:00
parent c8ac60144f
commit d12c157b66
4 changed files with 174 additions and 42 deletions

View File

@@ -11,7 +11,7 @@ from models.translation.translation_languages import translation_lang
from models.translation.translation_utils import ctranslate2_weights
from models.transcription.transcription_languages import transcription_lang
from models.transcription.transcription_whisper import _MODELS as whisper_models
from utils import errorLogging
from utils import errorLogging, validateDictStructure
json_serializable_vars = {}
def json_serializable(var_name):
@@ -139,21 +139,42 @@ class Config:
def SEND_MESSAGE_BUTTON_TYPE_LIST(self):
return self._SEND_MESSAGE_BUTTON_TYPE_LIST
@property
def SEND_MESSAGE_FORMAT(self):
return self._SEND_MESSAGE_FORMAT
@property
@json_serializable('SEND_MESSAGE_FORMAT_PARTS')
def SEND_MESSAGE_FORMAT_PARTS(self):
return self._SEND_MESSAGE_FORMAT_PARTS
@SEND_MESSAGE_FORMAT_PARTS.setter
def SEND_MESSAGE_FORMAT_PARTS(self, value):
if isinstance(value, dict):
valid_parts = {
"message": {"prefix": str, "suffix": str},
"separator": str,
"translation": {"prefix": str, "separator": str, "suffix": str},
"translation_first": bool
}
if validateDictStructure(value, valid_parts):
self._SEND_MESSAGE_FORMAT_PARTS = value
self.saveConfig(inspect.currentframe().f_code.co_name, value)
@property
def SEND_MESSAGE_FORMAT_WITH_T(self):
return self._SEND_MESSAGE_FORMAT_WITH_T
@json_serializable('RECEIVED_MESSAGE_FORMAT_PARTS')
def RECEIVED_MESSAGE_FORMAT_PARTS(self):
return self._RECEIVED_MESSAGE_FORMAT_PARTS
@property
def RECEIVED_MESSAGE_FORMAT(self):
return self._RECEIVED_MESSAGE_FORMAT
@property
def RECEIVED_MESSAGE_FORMAT_WITH_T(self):
return self._RECEIVED_MESSAGE_FORMAT_WITH_T
@RECEIVED_MESSAGE_FORMAT_PARTS.setter
def RECEIVED_MESSAGE_FORMAT_PARTS(self, value):
if isinstance(value, dict):
valid_parts = {
"message": {"prefix": str, "suffix": str},
"separator": str,
"translation": {"prefix": str, "separator": str, "suffix": str},
"translation_first": bool
}
if validateDictStructure(value, valid_parts):
self._RECEIVED_MESSAGE_FORMAT_PARTS = value
self.saveConfig(inspect.currentframe().f_code.co_name, value)
# Read Write
@property
@@ -997,7 +1018,6 @@ class Config:
self._WEBSOCKET_PORT = value
self.saveConfig(inspect.currentframe().f_code.co_name, value)
def init_config(self):
# Read Only
self._VERSION = "3.2.1"
@@ -1032,10 +1052,32 @@ class Config:
self._SELECTABLE_COMPUTE_DEVICE_LIST.append({"device":"cuda", "device_index": i, "device_name": torch.cuda.get_device_name(i)})
self._SELECTABLE_COMPUTE_DEVICE_LIST.append({"device":"cpu", "device_index": 0, "device_name": "cpu"})
self._SEND_MESSAGE_BUTTON_TYPE_LIST = ["show", "hide", "show_and_disable_enter_key"]
self._SEND_MESSAGE_FORMAT = "[message]"
self._SEND_MESSAGE_FORMAT_WITH_T = "[message]\n[translation]"
self._RECEIVED_MESSAGE_FORMAT = "[message]"
self._RECEIVED_MESSAGE_FORMAT_WITH_T = "[message]\n[translation]"
self._SEND_MESSAGE_FORMAT_PARTS = {
"message": {
"prefix": "",
"suffix": ""
},
"separator": "\n",
"translation": {
"prefix": "",
"separator": "\n",
"suffix": ""
},
"translation_first": False,
}
self._RECEIVED_MESSAGE_FORMAT_PARTS = {
"message": {
"prefix": "",
"suffix": ""
},
"separator": "\n",
"translation": {
"prefix": "",
"separator": "\n",
"suffix": ""
},
"translation_first": False,
}
# Read Write
self._ENABLE_TRANSLATION = False

View File

@@ -306,11 +306,11 @@ class Controller:
if config.SEND_MESSAGE_TO_VRC is True:
if config.SEND_ONLY_TRANSLATED_MESSAGES is True:
if config.ENABLE_TRANSLATION is False:
osc_message = self.messageFormatter("SEND", "", [message])
osc_message = self.messageFormatter("SEND", [], message)
else:
osc_message = self.messageFormatter("SEND", "", translation)
osc_message = self.messageFormatter("SEND", translation, "")
else:
osc_message = self.messageFormatter("SEND", translation, [message])
osc_message = self.messageFormatter("SEND", translation, message)
model.oscSendMessage(osc_message)
self.run(
@@ -470,7 +470,13 @@ class Controller:
model.updateOverlayLargeLog(overlay_image)
if config.SEND_RECEIVED_MESSAGE_TO_VRC is True:
osc_message = self.messageFormatter("RECEIVED", translation, [message])
if config.SEND_ONLY_TRANSLATED_MESSAGES is True:
if config.ENABLE_TRANSLATION is False:
osc_message = self.messageFormatter("RECEIVED", [], message)
else:
osc_message = self.messageFormatter("RECEIVED", translation, "")
else:
osc_message = self.messageFormatter("RECEIVED", translation, message)
model.oscSendMessage(osc_message)
# update textbox message log (Received)
@@ -573,11 +579,11 @@ class Controller:
if config.SEND_MESSAGE_TO_VRC is True:
if config.SEND_ONLY_TRANSLATED_MESSAGES is True:
if config.ENABLE_TRANSLATION is False:
osc_message = self.messageFormatter("SEND", "", [message])
osc_message = self.messageFormatter("SEND", [], message)
else:
osc_message = self.messageFormatter("SEND", "", translation)
osc_message = self.messageFormatter("SEND", translation, "")
else:
osc_message = self.messageFormatter("SEND", translation, [message])
osc_message = self.messageFormatter("SEND", translation, message)
model.oscSendMessage(osc_message)
if config.OVERLAY_LARGE_LOG is True:
@@ -1450,6 +1456,24 @@ class Controller:
config.WHISPER_WEIGHT_TYPE = str(data)
return {"status":200, "result": config.WHISPER_WEIGHT_TYPE}
@staticmethod
def getSendMessageFormatParts(*args, **kwargs) -> dict:
return {"status":200, "result":config.SEND_MESSAGE_FORMAT_PARTS}
@staticmethod
def setSendMessageFormatParts(data, *args, **kwargs) -> dict:
config.SEND_MESSAGE_FORMAT_PARTS = str(data)
return {"status":200, "result":config.SEND_MESSAGE_FORMAT_PARTS}
@staticmethod
def getReceivedMessageFormatParts(*args, **kwargs) -> dict:
return {"status":200, "result":config.RECEIVED_MESSAGE_FORMAT_PARTS}
@staticmethod
def setReceivedMessageFormatParts(data, *args, **kwargs) -> dict:
config.RECEIVED_MESSAGE_FORMAT_PARTS = str(data)
return {"status":200, "result":config.RECEIVED_MESSAGE_FORMAT_PARTS}
@staticmethod
def getAutoClearMessageBox(*args, **kwargs) -> dict:
return {"status":200, "result":config.AUTO_CLEAR_MESSAGE_BOX}
@@ -1769,21 +1793,27 @@ class Controller:
return {"status":200, "result":True}
@staticmethod
def messageFormatter(format_type:str, translation:list, message:list) -> str:
def messageFormatter(format_type:str, translation:list, message:str) -> str:
if format_type == "RECEIVED":
FORMAT_WITH_T = config.RECEIVED_MESSAGE_FORMAT_WITH_T
FORMAT = config.RECEIVED_MESSAGE_FORMAT
format_parts = config.RECEIVED_MESSAGE_FORMAT_PARTS
elif format_type == "SEND":
FORMAT_WITH_T = config.SEND_MESSAGE_FORMAT_WITH_T
FORMAT = config.SEND_MESSAGE_FORMAT
format_parts = config.SEND_MESSAGE_FORMAT_PARTS
else:
raise ValueError("format_type is not found", format_type)
if len(translation) > 0:
osc_message = FORMAT_WITH_T.replace("[message]", "\n".join(message))
osc_message = osc_message.replace("[translation]", "\n".join(translation))
message_part = format_parts["message"]["prefix"] + message + format_parts["message"]["suffix"]
translation_part = format_parts["translation"]["prefix"] + format_parts["translation"]["separator"].join(translation) + format_parts["translation"]["suffix"]
if len(translation) > 0 and message != "":
# 翻訳とメッセージの順序を決定
if format_parts["translation_first"]:
osc_message = translation_part + format_parts["separator"] + message_part
else:
osc_message = message_part + format_parts["separator"] + translation_part
elif len(translation) > 0 and message == "":
osc_message = translation_part
else:
osc_message = FORMAT.replace("[message]", "\n".join(message))
osc_message = message_part
return osc_message
def changeToCTranslate2Process(self) -> None:

View File

@@ -285,6 +285,11 @@ mapping = {
"/set/disable/overlay_show_only_translated_messages": {"status": True, "variable":controller.setDisableOverlayShowOnlyTranslatedMessages},
# Others
"/get/data/send_message_format_parts": {"status": True, "variable":controller.getSendMessageFormatParts},
"/set/data/send_message_format_parts": {"status": True, "variable":controller.setSendMessageFormatParts},
"/get/data/received_message_format_parts": {"status": True, "variable":controller.getReceivedMessageFormatParts},
"/set/data/received_message_format_parts": {"status": True, "variable":controller.setReceivedMessageFormatParts},
"/get/data/auto_clear_message_box": {"status": True, "variable":controller.getAutoClearMessageBox},
"/set/enable/auto_clear_message_box": {"status": True, "variable":controller.setEnableAutoClearMessageBox},
"/set/disable/auto_clear_message_box": {"status": True, "variable":controller.setDisableAutoClearMessageBox},
@@ -591,14 +596,34 @@ if __name__ == "__main__":
"display_duration": 5,
"fadeout_duration": 0.5,
}
case "/set/data/send_message_format":
data = "[message]"
case "/set/data/send_message_format_with_t":
data = "[message]([translation])"
case "/set/data/received_message_format":
data = "[message]"
case "/set/data/received_message_format_with_t":
data = "[message]([translation])"
case "/set/data/send_message_format_parts":
data = {
"message": {
"prefix": "",
"suffix": ""
},
"between_separator": "\n",
"translation": {
"prefix": "(",
"separator": "\\",
"suffix": ")"
},
"translation_first": False,
}
case "/set/data/received_message_format_parts":
data = {
"message": {
"prefix": "",
"suffix": ""
},
"between_separator": "\n",
"translation": {
"prefix": "(",
"separator": "\\",
"suffix": ")"
},
"translation_first": True,
}
case "/set/data/osc_ip_address":
data = "127.0.0.1"
case "/set/data/osc_port":

View File

@@ -10,6 +10,41 @@ import requests
import ipaddress
import socket
def validateDictStructure(data: dict, structure: dict) -> bool:
"""
辞書とその期待される構造(型)が完全に一致するかを判別する関数
Args:
data (dict): 検証対象の辞書
structure (dict): 期待される構造を定義した辞書値には型str, int, bool等や入れ子の辞書を指定
Returns:
bool: 構造が完全に一致する場合True、そうでなければFalse
"""
if not isinstance(data, dict) or not isinstance(structure, dict):
return False
# キーの数と名前が完全に一致するかチェック
if set(data.keys()) != set(structure.keys()):
return False
# 各キーの値の型または構造をチェック
for key, expected_type_or_structure in structure.items():
if key not in data:
return False
value = data[key]
# 期待される型が辞書の場合(入れ子構造)
if isinstance(expected_type_or_structure, dict):
# 再帰的に検証(多重入れ子に対応)
if not validateDictStructure(value, expected_type_or_structure):
return False
# 期待される型が型オブジェクトの場合
else:
if not isinstance(value, expected_type_or_structure):
return False
return True
def isConnectedNetwork(url="http://www.google.com", timeout=3) -> bool:
try:
response = requests.get(url, timeout=timeout)