Controllerクラスの設定メソッドを修正し、重複した設定を防止する条件を追加。テストケースを更新し、ON/OFF状態のランダムアクセスと連続テストを強化。

This commit is contained in:
misyaguziya
2025-09-23 16:56:11 +09:00
parent 1689a45e3e
commit a4656a5081
2 changed files with 179 additions and 105 deletions

View File

@@ -795,6 +795,7 @@ class Controller:
@staticmethod
def setEnableTranslation(*args, **kwargs) -> dict:
if config.ENABLE_TRANSLATION is False:
if model.isLoadedCTranslate2Model() is False:
model.changeTranslatorCTranslate2Model()
config.ENABLE_TRANSLATION = True
@@ -802,16 +803,19 @@ class Controller:
@staticmethod
def setDisableTranslation(*args, **kwargs) -> dict:
if config.ENABLE_TRANSLATION is True:
config.ENABLE_TRANSLATION = False
return {"status":200, "result":config.ENABLE_TRANSLATION}
@staticmethod
def setEnableForeground(*args, **kwargs) -> dict:
if config.ENABLE_FOREGROUND is False:
config.ENABLE_FOREGROUND = True
return {"status":200, "result":config.ENABLE_FOREGROUND}
@staticmethod
def setDisableForeground(*args, **kwargs) -> dict:
if config.ENABLE_FOREGROUND is True:
config.ENABLE_FOREGROUND = False
return {"status":200, "result":config.ENABLE_FOREGROUND}
@@ -906,11 +910,13 @@ class Controller:
@staticmethod
def setEnableConvertMessageToRomaji(*args, **kwargs) -> dict:
if config.CONVERT_MESSAGE_TO_ROMAJI is False:
config.CONVERT_MESSAGE_TO_ROMAJI = True
return {"status":200, "result":config.CONVERT_MESSAGE_TO_ROMAJI}
@staticmethod
def setDisableConvertMessageToRomaji(*args, **kwargs) -> dict:
if config.CONVERT_MESSAGE_TO_ROMAJI is True:
config.CONVERT_MESSAGE_TO_ROMAJI = False
return {"status":200, "result":config.CONVERT_MESSAGE_TO_ROMAJI}
@@ -920,11 +926,13 @@ class Controller:
@staticmethod
def setEnableConvertMessageToHiragana(*args, **kwargs) -> dict:
if config.CONVERT_MESSAGE_TO_HIRAGANA is False:
config.CONVERT_MESSAGE_TO_HIRAGANA = True
return {"status":200, "result":config.CONVERT_MESSAGE_TO_HIRAGANA}
@staticmethod
def setDisableConvertMessageToHiragana(*args, **kwargs) -> dict:
if config.CONVERT_MESSAGE_TO_HIRAGANA is True:
config.CONVERT_MESSAGE_TO_HIRAGANA = False
return {"status":200, "result":config.CONVERT_MESSAGE_TO_HIRAGANA}
@@ -934,11 +942,13 @@ class Controller:
@staticmethod
def setEnableMainWindowSidebarCompactMode(*args, **kwargs) -> dict:
if config.MAIN_WINDOW_SIDEBAR_COMPACT_MODE is False:
config.MAIN_WINDOW_SIDEBAR_COMPACT_MODE = True
return {"status":200, "result":config.MAIN_WINDOW_SIDEBAR_COMPACT_MODE}
@staticmethod
def setDisableMainWindowSidebarCompactMode(*args, **kwargs) -> dict:
if config.MAIN_WINDOW_SIDEBAR_COMPACT_MODE is True:
config.MAIN_WINDOW_SIDEBAR_COMPACT_MODE = False
return {"status":200, "result":config.MAIN_WINDOW_SIDEBAR_COMPACT_MODE}
@@ -993,11 +1003,13 @@ class Controller:
@staticmethod
def setEnableShowResendButton(*args, **kwargs) -> dict:
if not config.SHOW_RESEND_BUTTON:
config.SHOW_RESEND_BUTTON = True
return {"status":200, "result":config.SHOW_RESEND_BUTTON}
@staticmethod
def setDisableShowResendButton(*args, **kwargs) -> dict:
if config.SHOW_RESEND_BUTTON is True:
config.SHOW_RESEND_BUTTON = False
return {"status":200, "result":config.SHOW_RESEND_BUTTON}
@@ -1033,15 +1045,17 @@ class Controller:
return {"status":200, "result":config.AUTO_MIC_SELECT}
def setEnableAutoMicSelect(self, *args, **kwargs) -> dict:
config.AUTO_MIC_SELECT = True
if config.AUTO_MIC_SELECT is False:
device_manager.setCallbackProcessBeforeUpdateDevices(self.stopAccessDevices)
device_manager.setCallbackDefaultMicDevice(self.updateSelectedMicDevice)
device_manager.setCallbackProcessAfterUpdateDevices(self.restartAccessDevices)
device_manager.forceUpdateAndSetMicDevices()
config.AUTO_MIC_SELECT = True
return {"status":200, "result":config.AUTO_MIC_SELECT}
@staticmethod
def setDisableAutoMicSelect(*args, **kwargs) -> dict:
if config.AUTO_MIC_SELECT is True:
device_manager.clearCallbackProcessBeforeUpdateDevices()
device_manager.clearCallbackDefaultMicDevice()
device_manager.clearCallbackProcessAfterUpdateDevices()
@@ -1108,11 +1122,13 @@ class Controller:
@staticmethod
def setEnableMicAutomaticThreshold(*args, **kwargs) -> dict:
if config.MIC_AUTOMATIC_THRESHOLD is False:
config.MIC_AUTOMATIC_THRESHOLD = True
return {"status":200, "result":config.MIC_AUTOMATIC_THRESHOLD}
@staticmethod
def setDisableMicAutomaticThreshold(*args, **kwargs) -> dict:
if config.MIC_AUTOMATIC_THRESHOLD is True:
config.MIC_AUTOMATIC_THRESHOLD = False
return {"status":200, "result":config.MIC_AUTOMATIC_THRESHOLD}
@@ -1226,16 +1242,17 @@ class Controller:
return {"status":200, "result":config.AUTO_SPEAKER_SELECT}
def setEnableAutoSpeakerSelect(self, *args, **kwargs) -> dict:
config.AUTO_SPEAKER_SELECT = True
if config.AUTO_SPEAKER_SELECT is False:
device_manager.setCallbackProcessBeforeUpdateDevices(self.stopAccessDevices)
device_manager.setCallbackDefaultSpeakerDevice(self.updateSelectedSpeakerDevice)
device_manager.setCallbackProcessAfterUpdateDevices(self.restartAccessDevices)
device_manager.forceUpdateAndSetSpeakerDevices()
config.AUTO_SPEAKER_SELECT = True
return {"status":200, "result":config.AUTO_SPEAKER_SELECT}
@staticmethod
def setDisableAutoSpeakerSelect(*args, **kwargs) -> dict:
if config.AUTO_SPEAKER_SELECT is True:
device_manager.clearCallbackProcessBeforeUpdateDevices()
device_manager.clearCallbackDefaultSpeakerDevice()
device_manager.clearCallbackProcessAfterUpdateDevices()
@@ -1285,11 +1302,13 @@ class Controller:
@staticmethod
def setEnableSpeakerAutomaticThreshold(*args, **kwargs) -> dict:
if config.SPEAKER_AUTOMATIC_THRESHOLD is False:
config.SPEAKER_AUTOMATIC_THRESHOLD = True
return {"status":200, "result":config.SPEAKER_AUTOMATIC_THRESHOLD}
@staticmethod
def setDisableSpeakerAutomaticThreshold(*args, **kwargs) -> dict:
if config.SPEAKER_AUTOMATIC_THRESHOLD is True:
config.SPEAKER_AUTOMATIC_THRESHOLD = False
return {"status":200, "result":config.SPEAKER_AUTOMATIC_THRESHOLD}
@@ -1459,11 +1478,13 @@ class Controller:
@staticmethod
def setEnableNotificationVrcSfx(*args, **kwargs) -> dict:
if config.NOTIFICATION_VRC_SFX is False:
config.NOTIFICATION_VRC_SFX = True
return {"status":200, "result":config.NOTIFICATION_VRC_SFX}
@staticmethod
def setDisableNotificationVrcSfx(*args, **kwargs) -> dict:
if config.NOTIFICATION_VRC_SFX is True:
config.NOTIFICATION_VRC_SFX = False
return {"status":200, "result":config.NOTIFICATION_VRC_SFX}
@@ -1571,11 +1592,13 @@ class Controller:
@staticmethod
def setEnableAutoClearMessageBox(*args, **kwargs) -> dict:
if config.AUTO_CLEAR_MESSAGE_BOX is False:
config.AUTO_CLEAR_MESSAGE_BOX = True
return {"status":200, "result":config.AUTO_CLEAR_MESSAGE_BOX}
@staticmethod
def setDisableAutoClearMessageBox(*args, **kwargs) -> dict:
if config.AUTO_CLEAR_MESSAGE_BOX is True:
config.AUTO_CLEAR_MESSAGE_BOX = False
return {"status":200, "result":config.AUTO_CLEAR_MESSAGE_BOX}
@@ -1585,11 +1608,13 @@ class Controller:
@staticmethod
def setEnableSendOnlyTranslatedMessages(*args, **kwargs) -> dict:
if config.SEND_ONLY_TRANSLATED_MESSAGES is False:
config.SEND_ONLY_TRANSLATED_MESSAGES = True
return {"status":200, "result":config.SEND_ONLY_TRANSLATED_MESSAGES}
@staticmethod
def setDisableSendOnlyTranslatedMessages(*args, **kwargs) -> dict:
if config.SEND_ONLY_TRANSLATED_MESSAGES is True:
config.SEND_ONLY_TRANSLATED_MESSAGES = False
return {"status":200, "result":config.SEND_ONLY_TRANSLATED_MESSAGES}
@@ -1599,17 +1624,19 @@ class Controller:
@staticmethod
def setEnableOverlaySmallLog(*args, **kwargs) -> dict:
config.OVERLAY_SMALL_LOG = True
if config.OVERLAY_SMALL_LOG is False:
if config.OVERLAY_LARGE_LOG is False:
model.startOverlay()
config.OVERLAY_SMALL_LOG = True
return {"status":200, "result":config.OVERLAY_SMALL_LOG}
@staticmethod
def setDisableOverlaySmallLog(*args, **kwargs) -> dict:
config.OVERLAY_SMALL_LOG = False
if config.OVERLAY_SMALL_LOG is True:
model.clearOverlayImageSmallLog()
if config.OVERLAY_LARGE_LOG is False:
model.shutdownOverlay()
config.OVERLAY_SMALL_LOG = False
return {"status":200, "result":config.OVERLAY_SMALL_LOG}
@staticmethod
@@ -1628,17 +1655,19 @@ class Controller:
@staticmethod
def setEnableOverlayLargeLog(*args, **kwargs) -> dict:
config.OVERLAY_LARGE_LOG = True
if config.OVERLAY_LARGE_LOG is False:
if config.OVERLAY_SMALL_LOG is False:
model.startOverlay()
config.OVERLAY_LARGE_LOG = True
return {"status":200, "result":config.OVERLAY_LARGE_LOG}
@staticmethod
def setDisableOverlayLargeLog(*args, **kwargs) -> dict:
config.OVERLAY_LARGE_LOG = False
if config.OVERLAY_LARGE_LOG is True:
model.clearOverlayImageLargeLog()
if config.OVERLAY_SMALL_LOG is False:
model.shutdownOverlay()
config.OVERLAY_LARGE_LOG = False
return {"status":200, "result":config.OVERLAY_LARGE_LOG}
@staticmethod
@@ -1657,11 +1686,13 @@ class Controller:
@staticmethod
def setEnableOverlayShowOnlyTranslatedMessages(*args, **kwargs) -> dict:
if config.OVERLAY_SHOW_ONLY_TRANSLATED_MESSAGES is False:
config.OVERLAY_SHOW_ONLY_TRANSLATED_MESSAGES = True
return {"status":200, "result":config.OVERLAY_SHOW_ONLY_TRANSLATED_MESSAGES}
@staticmethod
def setDisableOverlayShowOnlyTranslatedMessages(*args, **kwargs) -> dict:
if config.OVERLAY_SHOW_ONLY_TRANSLATED_MESSAGES is True:
config.OVERLAY_SHOW_ONLY_TRANSLATED_MESSAGES = False
return {"status":200, "result":config.OVERLAY_SHOW_ONLY_TRANSLATED_MESSAGES}
@@ -1671,11 +1702,13 @@ class Controller:
@staticmethod
def setEnableSendMessageToVrc(*args, **kwargs) -> dict:
if config.SEND_MESSAGE_TO_VRC is False:
config.SEND_MESSAGE_TO_VRC = True
return {"status":200, "result":config.SEND_MESSAGE_TO_VRC}
@staticmethod
def setDisableSendMessageToVrc(*args, **kwargs) -> dict:
if config.SEND_MESSAGE_TO_VRC is True:
config.SEND_MESSAGE_TO_VRC = False
return {"status":200, "result":config.SEND_MESSAGE_TO_VRC}
@@ -1685,11 +1718,13 @@ class Controller:
@staticmethod
def setEnableSendReceivedMessageToVrc(*args, **kwargs) -> dict:
if config.SEND_RECEIVED_MESSAGE_TO_VRC is False:
config.SEND_RECEIVED_MESSAGE_TO_VRC = True
return {"status":200, "result":config.SEND_RECEIVED_MESSAGE_TO_VRC}
@staticmethod
def setDisableSendReceivedMessageToVrc(*args, **kwargs) -> dict:
if config.SEND_RECEIVED_MESSAGE_TO_VRC is True:
config.SEND_RECEIVED_MESSAGE_TO_VRC = False
return {"status":200, "result":config.SEND_RECEIVED_MESSAGE_TO_VRC}
@@ -1717,6 +1752,7 @@ class Controller:
@staticmethod
def setEnableVrcMicMuteSync(*args, **kwargs) -> dict:
if config.VRC_MIC_MUTE_SYNC is False:
if model.getIsOscQueryEnabled() is True:
config.VRC_MIC_MUTE_SYNC = True
model.setMuteSelfStatus()
@@ -1730,30 +1766,37 @@ class Controller:
"data": config.VRC_MIC_MUTE_SYNC
}
}
else:
response = {"status":200, "result":config.VRC_MIC_MUTE_SYNC}
return response
@staticmethod
def setDisableVrcMicMuteSync(*args, **kwargs) -> dict:
if config.VRC_MIC_MUTE_SYNC is True:
config.VRC_MIC_MUTE_SYNC = False
model.changeMicTranscriptStatus()
return {"status":200, "result":config.VRC_MIC_MUTE_SYNC}
def setEnableCheckSpeakerThreshold(self, *args, **kwargs) -> dict:
if config.ENABLE_CHECK_ENERGY_RECEIVE is False:
self.startThreadingCheckSpeakerEnergy()
config.ENABLE_CHECK_ENERGY_RECEIVE = True
return {"status":200, "result":config.ENABLE_CHECK_ENERGY_RECEIVE}
def setDisableCheckSpeakerThreshold(self, *args, **kwargs) -> dict:
if config.ENABLE_CHECK_ENERGY_RECEIVE is True:
self.stopThreadingCheckSpeakerEnergy()
config.ENABLE_CHECK_ENERGY_RECEIVE = False
return {"status":200, "result":config.ENABLE_CHECK_ENERGY_RECEIVE}
def setEnableCheckMicThreshold(self, *args, **kwargs) -> dict:
if config.ENABLE_CHECK_ENERGY_SEND is False:
self.startThreadingCheckMicEnergy()
config.ENABLE_CHECK_ENERGY_SEND = True
return {"status":200, "result":config.ENABLE_CHECK_ENERGY_SEND}
def setDisableCheckMicThreshold(self, *args, **kwargs) -> dict:
if config.ENABLE_CHECK_ENERGY_SEND is True:
self.stopThreadingCheckMicEnergy()
config.ENABLE_CHECK_ENERGY_SEND = False
return {"status":200, "result":config.ENABLE_CHECK_ENERGY_SEND}
@@ -1769,21 +1812,25 @@ class Controller:
return {"status":200, "result":True}
def setEnableTranscriptionSend(self, *args, **kwargs) -> dict:
if config.ENABLE_TRANSCRIPTION_SEND is False:
self.startThreadingTranscriptionSendMessage()
config.ENABLE_TRANSCRIPTION_SEND = True
return {"status":200, "result":config.ENABLE_TRANSCRIPTION_SEND}
def setDisableTranscriptionSend(self, *args, **kwargs) -> dict:
if config.ENABLE_TRANSCRIPTION_SEND is True:
self.stopThreadingTranscriptionSendMessage()
config.ENABLE_TRANSCRIPTION_SEND = False
return {"status":200, "result":config.ENABLE_TRANSCRIPTION_SEND}
def setEnableTranscriptionReceive(self, *args, **kwargs) -> dict:
if config.ENABLE_TRANSCRIPTION_RECEIVE is False:
self.startThreadingTranscriptionReceiveMessage()
config.ENABLE_TRANSCRIPTION_RECEIVE = True
return {"status":200, "result":config.ENABLE_TRANSCRIPTION_RECEIVE}
def setDisableTranscriptionReceive(self, *args, **kwargs) -> dict:
if config.ENABLE_TRANSCRIPTION_RECEIVE is True:
self.stopThreadingTranscriptionReceiveMessage()
config.ENABLE_TRANSCRIPTION_RECEIVE = False
return {"status":200, "result":config.ENABLE_TRANSCRIPTION_RECEIVE}

View File

@@ -73,18 +73,18 @@ class TestMainloop():
"/set/disable/convert_message_to_romaji",
"/set/enable/convert_message_to_hiragana",
"/set/disable/convert_message_to_hiragana",
# "/set/enable/auto_mic_select",
# "/set/disable/auto_mic_select",
"/set/enable/auto_mic_select",
"/set/disable/auto_mic_select",
"/set/enable/mic_automatic_threshold",
"/set/disable/mic_automatic_threshold",
# "/set/enable/check_mic_threshold",
# "/set/disable/check_mic_threshold",
# "/set/enable/auto_speaker_select",
# "/set/disable/auto_speaker_select",
"/set/enable/check_mic_threshold",
"/set/disable/check_mic_threshold",
"/set/enable/auto_speaker_select",
"/set/disable/auto_speaker_select",
"/set/enable/speaker_automatic_threshold",
"/set/disable/speaker_automatic_threshold",
# "/set/enable/check_speaker_threshold",
# "/set/disable/check_speaker_threshold",
"/set/enable/check_speaker_threshold",
"/set/disable/check_speaker_threshold",
"/set/enable/overlay_small_log",
"/set/disable/overlay_small_log",
"/set/enable/overlay_large_log",
@@ -251,12 +251,26 @@ class TestMainloop():
print(f"\t -> {Color.RED}[ERROR]{Color.RESET} Status: {status}, Result: {result}, Expected: {expected_result}")
pprint.pprint(self.config_dict)
break
time.sleep(0.2)
# 最後にすべてOFFにして終了
for endpoint in self.validity_endpoints:
if endpoint.startswith("/set/disable/"):
result, status = self.main.handleRequest(endpoint, None)
time.sleep(0.2)
print("----ON/OFFでのランダムアクセスのテスト終了----")
def test_endpoints_continuous(self):
def test_endpoints_on_off_continuous(self):
print("----連続テスト----")
# endpoints = ["/set/enable/websocket_server", "/set/disable/websocket_server"]
endpoints = ["/set/enable/transcription_receive", "/set/disable/transcription_receive"]
endpoints = [
"/set/enable/translation",
"/set/disable/translation",
"/set/enable/transcription_send",
"/set/disable/transcription_send",
"/set/enable/transcription_receive",
"/set/disable/transcription_receive",
]
for i in range(1000):
endpoint = random.choice(endpoints)
print(f"No.{i:04} Testing endpoint: {endpoint}", end="", flush=True)
@@ -280,8 +294,21 @@ class TestMainloop():
print(f"\t -> {Color.RED}[ERROR]{Color.RESET} Status: {status}, Result: {result}, Expected: {expected_result}")
pprint.pprint(self.config_dict)
break
time.sleep(0.2)
# 最後にすべてOFFにして終了
for endpoint in self.validity_endpoints:
if endpoint.startswith("/set/disable/"):
result, status = self.main.handleRequest(endpoint, None)
time.sleep(0.2)
print("----連続テスト終了----")
if __name__ == "__main__":
try:
test = TestMainloop()
test.test_endpoints_continuous()
test.test_endpoints_on_off_random()
# test.test_endpoints_continuous()
except KeyboardInterrupt:
print("Interrupted by user, shutting down...")
except Exception as e:
print(f"An error occurred: {e}")