diff --git a/src-python/backend_test.py b/src-python/backend_test.py new file mode 100644 index 00000000..b0ca93be --- /dev/null +++ b/src-python/backend_test.py @@ -0,0 +1,674 @@ +# 初期化のため、config.jsonの削除 +import os +import time +import random +if os.path.exists("config.json"): + os.remove("config.json") + +from mainloop import main_instance + +class Color: + BLACK = '\033[30m'#(文字)黒 + RED = '\033[31m'#(文字)赤 + GREEN = '\033[32m'#(文字)緑 + YELLOW = '\033[33m'#(文字)黄 + BLUE = '\033[34m'#(文字)青 + MAGENTA = '\033[35m'#(文字)マゼンタ + CYAN = '\033[36m'#(文字)シアン + WHITE = '\033[37m'#(文字)白 + COLOR_DEFAULT = '\033[39m'#文字色をデフォルトに戻す + BOLD = '\033[1m'#太字 + UNDERLINE = '\033[4m'#下線 + INVISIBLE = '\033[08m'#不可視 + REVERCE = '\033[07m'#文字色と背景色を反転 + BG_BLACK = '\033[40m'#(背景)黒 + BG_RED = '\033[41m'#(背景)赤 + BG_GREEN = '\033[42m'#(背景)緑 + BG_YELLOW = '\033[43m'#(背景)黄 + BG_BLUE = '\033[44m'#(背景)青 + BG_MAGENTA = '\033[45m'#(背景)マゼンタ + BG_CYAN = '\033[46m'#(背景)シアン + BG_WHITE = '\033[47m'#(背景)白 + BG_DEFAULT = '\033[49m'#背景色をデフォルトに戻す + RESET = '\033[0m'#全てリセット + +class TestMainloop(): + def __init__(self): + self.main = main_instance + self.main.startReceiver() + self.main.startHandler() + + def stop_main(): + pass + self.main.controller.setWatchdogCallback(stop_main) + self.main.controller.init() + + # mappingのすべてのstatusをTrueにする + for key in self.main.mapping.keys(): + self.main.mapping[key]["status"] = True + + self.config_dict = {} + for endpoint in self.main.mapping.keys(): + if endpoint.startswith("/get/data/"): + self.config_dict[endpoint.split("/")[-1]], _ = self.main.handleRequest(endpoint, None) + elif endpoint.startswith("/set/disable/"): + self.config_dict[endpoint.split("/")[-1]], _ = self.main.handleRequest(endpoint, None) + print(self.config_dict, flush=True) + + self.validity_endpoints = [] + for endpoint in self.main.mapping.keys(): + if endpoint.startswith("/set/enable/") or endpoint.startswith("/set/disable/"): + self.validity_endpoints.append(endpoint) + + self.set_data_endpoints = [] + for endpoint in self.main.mapping.keys(): + if endpoint.startswith("/set/data/"): + self.set_data_endpoints.append(endpoint) + + self.delete_data_endpoints = [] + for endpoint in self.main.mapping.keys(): + if endpoint.startswith("/delete/data/"): + self.delete_data_endpoints.append(endpoint) + + self.run_endpoints = [] + for endpoint in self.main.mapping.keys(): + if endpoint.startswith("/run/"): + self.run_endpoints.append(endpoint) + + self.test_results = {} + + def record_test_result(self, endpoint, status, result, expected_status): + """ + テスト結果を記録する + :param endpoint: テスト対象のエンドポイント + :param status: 実際のステータスコード + :param result: 実際の結果 + :param expected_status: 期待されるステータスコード + """ + self.test_results[endpoint] = { + "status": status, + "result": result, + "expected_status": expected_status, + "success": status in expected_status + } + + def test_endpoints_on_off_single(self, endpoint): + success = False + expected_status = [200] + if endpoint.startswith("/set/enable/"): + match endpoint: + case "/set/enable/websocket_server": + expected_status = [200, 400] + case _: + pass + + result, status = self.main.handleRequest(endpoint, None) + if status in expected_status: + if status == 200: + self.config_dict[endpoint.split("/")[-1]] = result + print(f"-> {Color.GREEN}[PASS]{Color.RESET} endpoint:{endpoint} Status: {status}, Result: {result}") + success = True + else: + print(f"-> {Color.RED}[ERROR]{Color.RESET} endpoint:{endpoint} Status: {status}, Result: {result}") + print(f"Current config_dict: {self.config_dict}") + elif endpoint.startswith("/set/disable/"): + result, status = self.main.handleRequest(endpoint, None) + if status in expected_status: + if status == 200: + self.config_dict[endpoint.split("/")[-1]] = result + print(f"-> {Color.GREEN}[PASS]{Color.RESET} endpoint:{endpoint} Status: {status}, Result: {result}") + success = True + else: + print(f"-> {Color.RED}[ERROR]{Color.RESET} endpoint:{endpoint} Status: {status}, Result: {result}") + print(f"Current config_dict: {self.config_dict}") + self.record_test_result(endpoint, status, result, expected_status) + return success + + def test_endpoints_on_off_all(self): + print("----ON/OFF系のエンドポイントのテスト----") + for endpoint in self.validity_endpoints: + print(f"Testing endpoint: {endpoint}", flush=True) + self.test_endpoints_on_off_single(endpoint) + print("----ON/OFF系のエンドポイントのテスト終了----") + + def test_endpoints_on_off_random(self): + print("----ON/OFFでのランダムアクセスのテスト----") + for i in range(1000): + endpoint = random.choice(self.validity_endpoints) + print(f"No.{i:04} Testing endpoint: {endpoint}", flush=True) + if self.test_endpoints_on_off_single(endpoint) is False: + break + + # 最後にすべて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_on_off_continuous(self): + print("----ON/OFF連続テスト----") + endpoints = ["/set/enable/websocket_server", "/set/disable/websocket_server"] + # 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}", flush=True) + if self.test_endpoints_on_off_single(endpoint) is False: + break + + # 最後にすべてOFFにして終了 + for endpoint in self.validity_endpoints: + if endpoint.startswith("/set/disable/"): + result, status = self.main.handleRequest(endpoint, None) + print("----ON/OFF連続テスト終了----") + + def test_set_data_endpoints_single(self, endpoint): + success = False + expected_status = [200] + match endpoint: + case "/set/data/selected_tab_no": + data = random.choice(["1", "2", "3"]) + case "/set/data/selected_translation_engines": + translation_engines = self.config_dict.get("translation_engines", None) + data = {} + for i in ["1", "2", "3"]: + data[i] = random.choice(translation_engines) + case "/set/data/selected_your_languages": + selectable_language_list = self.config_dict.get("selectable_language_list", None) + data = {} + for i in ["1", "2", "3"]: + data[i] = {} + data[i]["1"] = random.choice(selectable_language_list) | {"enable": True} + case "/set/data/selected_target_languages": + selectable_language_list = self.config_dict.get("selectable_language_list", None) + data = {} + for i in ["1", "2", "3"]: + data[i] = {} + for j in ["1", "2", "3"]: + data[i][j] = random.choice(selectable_language_list) | {"enable": random.choice([True, False])} + case "/set/data/selected_transcription_engine": + transcription_engines = self.config_dict.get("transcription_engines", None) + data = random.choice(transcription_engines) + case "/set/data/transparency": + data = random.randint(0, 100) + case "/set/data/ui_scaling": + data = random.randint(50, 200) + case "/set/data/textbox_ui_scaling": + data = random.randint(50, 200) + case "/set/data/message_box_ratio": + data = round(random.uniform(0.1, 0.9), 2) + case "/set/data/send_message_button_type": + data = random.choice(["show", "hide", "show_and_disable_enter_key"]) + case "/set/data/font_family": + data = random.choice(["Arial", "Verdana", "Times New Roman"]) + case "/set/data/ui_language": + data = random.choice(["en", "ja", "ko", "zh-Hant", "zh-Hans"]) + case "/set/data/main_window_geometry": + data = { + "x_pos": random.randint(0, 1920), + "y_pos": random.randint(0, 1080), + "width": random.randint(800, 1920), + "height": random.randint(600, 1080) + } + case "/set/data/selected_translation_compute_device": + data = random.choice(self.config_dict["translation_compute_device_list"]) + case "/set/data/selected_transcription_compute_device": + data = random.choice(self.config_dict["transcription_compute_device_list"]) + case "/set/data/ctranslate2_weight_type": + data = random.choice(list(self.config_dict["selectable_ctranslate2_weight_type_dict"].keys())) + case "/set/data/deepl_auth_key": + data = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + expected_status = [400] + case "/set/data/selected_mic_host": + data = random.choice(self.config_dict["mic_host_list"]) + case "/set/data/selected_mic_device": + data = random.choice(self.config_dict["mic_device_list"]) + case "/set/data/mic_threshold": + data = random.randint(-1000, 3000) + if 0 <= data <= 2000: + pass + else: + expected_status = [400] + case "/set/data/mic_record_timeout": + data = random.randint(-1, 10) + if 0 <= data <= self.config_dict["mic_phrase_timeout"]: + pass + else: + expected_status = [400] + case "/set/data/mic_phrase_timeout": + data = random.randint(-1, 10) + if self.config_dict["mic_record_timeout"] <= data: + pass + else: + expected_status = [400] + case "/set/data/mic_max_phrases": + data = random.randint(-1, 10) + if 0 <= data: + pass + else: + expected_status = [400] + case "/set/data/hotkeys": + data = { + 'toggle_vrct_visibility': None, + 'toggle_translation': None, + 'toggle_transcription_send': None, + 'toggle_transcription_receive': None + } + case "/set/data/plugins_status": + data = {plugin: random.choice([True, False]) for plugin in self.config_dict.get("plugins", [])} + case "/set/data/mic_avg_logprob": + data = random.uniform(-5, 0) + case "/set/data/mic_no_speech_prob": + data = random.uniform(0, 1) + case "/set/data/mic_word_filter": + data = random.choice( + [ + ["test_0_0", "test_0_1", "test_0_2", None], + ["test_1_0", "test_1_1", None], + ["test_2_0", None], + [None] + ] + ) + case "/set/data/selected_speaker_device": + data = random.choice(self.config_dict["speaker_device_list"]) + case "/set/data/speaker_threshold": + data = random.randint(-1000, 5000) + if 0 <= data <= 4000: + pass + else: + expected_status = [400] + case "/set/data/speaker_record_timeout": + data = random.randint(-1, 10) + if 0 <= data <= self.config_dict["speaker_phrase_timeout"]: + pass + else: + expected_status = [400] + case "/set/data/speaker_phrase_timeout": + data = random.randint(-1, 10) + if self.config_dict["speaker_record_timeout"] <= data: + pass + else: + expected_status = [400] + case "/set/data/speaker_max_phrases": + data = random.randint(-1, 10) + if 0 <= data: + pass + else: + expected_status = [400] + case "/set/data/speaker_avg_logprob": + data = random.uniform(-5, 0) + case "/set/data/speaker_no_speech_prob": + data = random.uniform(0, 1) + case "/set/data/whisper_weight_type": + data = random.choice([key for key, value in self.config_dict["selectable_whisper_weight_type_dict"].items() if value is True]) + case "/set/data/overlay_small_log_settings": + data = { + "x_pos": random.random(), + "y_pos": random.random(), + "z_pos": random.random(), + "x_rotation": random.random(), + "y_rotation": random.random(), + "z_rotation": random.random(), + "display_duration": random.randint(0, 100), + "fadeout_duration": random.randint(0, 100), + "opacity": random.random(), + "ui_scaling": random.random(), + "tracker": random.choice(["HMD", "LeftHand", "RightHand"]), + } + case "/set/data/overlay_large_log_settings": + data = { + "x_pos": random.random(), + "y_pos": random.random(), + "z_pos": random.random(), + "x_rotation": random.random(), + "y_rotation": random.random(), + "z_rotation": random.random(), + "display_duration": random.randint(0, 100), + "fadeout_duration": random.randint(0, 100), + "opacity": random.random(), + "ui_scaling": random.random(), + "tracker": random.choice(["HMD", "LeftHand", "RightHand"]), + } + case "/set/data/send_message_format_parts": + data = self.config_dict["send_message_format_parts"] + case "/set/data/received_message_format_parts": + data = self.config_dict["received_message_format_parts"] + case "/set/data/websocket_host": + data = random.choice(["127.0.0.1", "aaaaadwafasdsd", "0210.1564.845.0"]) + if data == "127.0.0.1": + expected_status = [200, 400] + else: + expected_status = [400] + case "/set/data/websocket_port": + data = random.randint(1024, 65535) + expected_status = [200, 400] + case "/set/data/osc_ip_address": + data = random.choice(["127.0.0.1", "aaaaadwafasdsd", "0210.1564.845.0"]) + if data == "127.0.0.1": + pass + else: + expected_status = [400] + case "/set/data/osc_port": + data = random.randint(1024, 65535) + case _: + data = None + expected_status = [404] + + if expected_status == [401]: + print(f"-> {Color.YELLOW}[SKIP]{Color.RESET} No test available for this endpoint: {endpoint}.") + self.record_test_result(endpoint, None, None, expected_status) # テスト結果を記録 + return success + elif expected_status == [404]: + print(f"-> {Color.RED}[ERROR]{Color.RESET} Unknown endpoint: {endpoint}.") + self.record_test_result(endpoint, None, None, expected_status) # テスト結果を記録 + return False + + if data is not None: + print(f"data: {data}", end=" ", flush=True) + result, status = self.main.handleRequest(endpoint, data) + if status in expected_status: + if status == 200: + self.config_dict[endpoint.split("/")[-1]] = result + print(f"-> {Color.GREEN}[PASS]{Color.RESET} endpoint:{endpoint} Status: {status}, Result: {result}") + success = True + else: + print(f"-> {Color.RED}[ERROR]{Color.RESET} endpoint:{endpoint} Status: {status}, Result: {result}") + print(f" Current config_dict: {self.config_dict}") + else: + print(f"-> {Color.YELLOW}[SKIP]{Color.RESET} No data to set for this endpoint: {endpoint}.") + success = True + self.record_test_result(endpoint, status, result if data is not None else None, expected_status) # テスト結果を記録 + return success + + def test_set_data_endpoints_all(self): + print("----データ設定系のエンドポイントのテスト----") + for endpoint in self.set_data_endpoints: + print(f"Testing endpoint: {endpoint}", end=" ", flush=True) + self.test_set_data_endpoints_single(endpoint) + print("----データ設定系のエンドポイントのテスト終了----") + + def test_run_endpoints_single(self, endpoint): + success = False + expected_status = [200] + match endpoint: + case "/run/send_message_box": + data_list = [ + { + "data": {"id":"000001", "message":"test"}, + "status": [200], + }, + { + # 英語 + "data": {"id":"000002", "message":"Hello World!"}, + "status": [200], + }, + { + # 日本語 + "data": {"id":"000003", "message":"こんにちわ 世界！"}, + "status": [200], + }, + { + # 韓国語 + "data": {"id":"000004", "message":"안녕하세요 세계!"}, + "status": [200], + }, + { + # 中国語 繁体字 + "data": {"id":"000005", "message":"你好，世界！"}, + "status": [200], + }, + ] + choice_data = random.choice(data_list) + data, expected_status = choice_data["data"], choice_data["status"] + case "/run/typing_message_box": + data = None + case "/run/stop_typing_message_box": + data = None + case "/run/send_text_overlay": + data = "test_overlay" + case "/run/swap_your_language_and_target_language": + data = None + case "/run/update_software": + data = None + expected_status = [401] # !!!Cant be tested here!!! + case "/run/update_cuda_software": + data = None + expected_status = [401] # !!!Cant be tested here!!! + case "/run/download_ctranslate2_weight": + data_list = random.choice(["small", "large"]) + data = random.choice(data_list) + expected_status = [401] # !!!Cant be tested here!!! + case "/run/download_whisper_weight": + data_list = [ + "tiny", "base", "small", "medium", + "large-v1", "large-v2", "large-v3", + "large-v3-turbo-int8", "large-v3-turbo" + ] + data = random.choice(data_list) + expected_status = [401] # !!!Cant be tested here!!! + case "/run/open_filepath_logs": + data = None + expected_status = [401] # !!!Cant be tested here!!! + case "/run/open_filepath_config_file": + data = None + expected_status = [401] # !!!Cant be tested here!!! + case "/run/feed_watchdog": + data = None + expected_status = [401] # !!!Cant be tested here!!! + case _: + data = None + expected_status = [404] + success = True + + if expected_status == [401]: + print(f"-> {Color.YELLOW}[SKIP]{Color.RESET} No test available for this endpoint: {endpoint}.") + self.record_test_result(endpoint, None, None, expected_status) # テスト結果を記録 + return success + elif expected_status == [404]: + print(f"-> {Color.RED}[ERROR]{Color.RESET} Unknown endpoint: {endpoint}.") + self.record_test_result(endpoint, None, None, expected_status) # テスト結果を記録 + return False + + result, status = self.main.handleRequest(endpoint, data) + if status == expected_status: + print(f"-> {Color.GREEN}[PASS]{Color.RESET} endpoint:{endpoint} Status: {status}, Result: {result}") + success = True + else: + print(f"-> {Color.RED}[ERROR]{Color.RESET} endpoint:{endpoint} Status: {status}, Result: {result}") + print(f"Current config_dict: {self.config_dict}") + self.record_test_result(endpoint, status, result, expected_status) # テスト結果を記録 + return success + + def test_run_endpoints_all(self): + print("----実行系のエンドポイントのテスト----") + for endpoint in self.run_endpoints: + print(f"Testing endpoint: {endpoint}", end=" ", flush=True) + self.test_run_endpoints_single(endpoint) + print("----実行系のエンドポイントのテスト終了----") + + def test_endpoints_all_random(self): + print("----すべてのエンドポイントのランダムアクセスのテスト----") + endpoint_types = [ + "validity", + "set_data", + "run", + "delete", + ] + + for i in range(10000): + endpoints_type = random.choice(endpoint_types) + match endpoints_type: + case "validity": + endpoint = random.choice(self.validity_endpoints) + print(f"No.{i:04} Testing endpoint: {endpoint}", flush=True) + if self.test_endpoints_on_off_single(endpoint) is False: + break + case "set_data": + endpoint = random.choice(self.set_data_endpoints) + print(f"No.{i:04} Testing endpoint: {endpoint}", flush=True) + if self.test_set_data_endpoints_single(endpoint) is False: + break + case "run": + endpoint = random.choice(self.run_endpoints) + print(f"No.{i:04} Testing endpoint: {endpoint}", flush=True) + if self.test_run_endpoints_single(endpoint) is False: + break + case "delete": + endpoint = random.choice(self.delete_data_endpoints) + print(f"No.{i:04} Testing endpoint: {endpoint}", flush=True) + if self.test_delete_data_endpoints_single(endpoint) is False: + break + + # 最後にすべてOFFにして終了 + for endpoint in self.validity_endpoints: + if endpoint.startswith("/set/disable/"): + _, _ = self.main.handleRequest(endpoint, None) + print("----すべてのエンドポイントのランダムアクセスのテスト終了----") + + def test_endpoints_specific_random(self): + print("----特定のエンドポイントのランダムアクセスのテスト----") + + self.validity_specific_endpoints = [ + "/set/enable/websocket_server", + "/set/disable/websocket_server", + ] + + self.set_data_specific_endpoints = [ + "/set/data/ctranslate2_weight_type", + "/set/data/websocket_host", + "/set/data/websocket_port", + "/set/data/osc_ip_address", + "/set/data/osc_port", + ] + + self.run_specific_endpoints = [] + self.delete_data_endpoints = [] + + endpoint_types = [ + "validity", + "set_data", + # "run", + # "delete", + ] + + for i in range(1000): + endpoints_type = random.choice(endpoint_types) + match endpoints_type: + case "validity": + endpoint = random.choice(self.validity_specific_endpoints) + print(f"No.{i:04} Testing endpoint: {endpoint}", flush=True) + if self.test_endpoints_on_off_single(endpoint) is False: + break + case "set_data": + endpoint = random.choice(self.set_data_specific_endpoints) + print(f"No.{i:04} Testing endpoint: {endpoint}", flush=True) + if self.test_set_data_endpoints_single(endpoint) is False: + break + case "run": + endpoint = random.choice(self.run_specific_endpoints) + print(f"No.{i:04} Testing endpoint: {endpoint}", flush=True) + if self.test_run_endpoints_single(endpoint) is False: + break + case "delete": + endpoint = random.choice(self.delete_data_endpoints) + print(f"No.{i:04} Testing endpoint: {endpoint}", flush=True) + if self.test_delete_data_endpoints_single(endpoint) is False: + break + + # 最後にすべてOFFにして終了 + for endpoint in self.validity_endpoints: + if endpoint.startswith("/set/disable/"): + _, _ = self.main.handleRequest(endpoint, None) + print("----特定のエンドポイントのランダムアクセスのテスト終了----") + + def test_delete_data_endpoints_single(self, endpoint): + success = False + expected_status = [200] + match endpoint: + case "/delete/data/deepl_auth_key": + data = None + case _: + data = None + expected_status = [404] + success = True + + if expected_status == [404]: + print(f"-> {Color.RED}[ERROR]{Color.RESET} Unknown endpoint: {endpoint}.") + self.record_test_result(endpoint, None, None, expected_status) # テスト結果を記録 + return False + + result, status = self.main.handleRequest(endpoint, data) + if status == expected_status: + print(f"-> {Color.GREEN}[PASS]{Color.RESET} endpoint:{endpoint} Status: {status}, Result: {result}") + success = True + else: + print(f"-> {Color.RED}[ERROR]{Color.RESET} endpoint:{endpoint} Status: {status}, Result: {result}") + print(f"Current config_dict: {self.config_dict}") + self.record_test_result(endpoint, status, result, expected_status) # テスト結果を記録 + return success + + def test_delete_data_endpoints_all(self): + print("----データ削除系のエンドポイントのテスト----") + for endpoint in self.delete_data_endpoints: + print(f"Testing endpoint: {endpoint}", flush=True) + self.test_delete_data_endpoints_single(endpoint) + print("----データ削除系のエンドポイントのテスト終了----") + + def generate_summary(self): + """ + テスト結果のサマリーを生成して表示する + """ + total_tests = len(self.test_results) + passed_tests = sum(1 for result in self.test_results.values() if result["success"]) + untested_tests = sum(1 for result in self.test_results.values() if result["expected_status"] == [401]) + invalid_tests = sum(1 for result in self.test_results.values() if result["expected_status"] == [404]) + failed_tests = total_tests - passed_tests - untested_tests - invalid_tests + + print("\n---- テスト結果のサマリー ----") + print(f"総テスト数: {total_tests}") + print(f"成功したテスト数: {passed_tests}") + print(f"失敗したテスト数: {failed_tests}") + print(f"テストをしなかったテスト数: {untested_tests}") + print(f"無効なテスト数: {invalid_tests}\n") + + if untested_tests > 0: + print("テストをしなかったテストの詳細:") + for endpoint, result in self.test_results.items(): + if result["expected_status"] == [401]: + print(f"- エンドポイント: {endpoint}") + print(f" ステータス: {result['status']}") + print(f" 結果: {result['result']}\n") + if failed_tests > 0: + print("失敗したテストの詳細:") + for endpoint, result in self.test_results.items(): + if result["success"] != [200]: + print(f"- エンドポイント: {endpoint}") + print(f" ステータス: {result['status']} (期待されるステータス: {result['expected_status']})") + print(f" 結果: {result['result']}\n") + print("---- サマリー終了 ----\n") + +if __name__ == "__main__": + import traceback + try: + test = TestMainloop() + test.test_endpoints_on_off_all() + test.test_set_data_endpoints_all() + test.test_run_endpoints_all() + test.test_delete_data_endpoints_all() + # test.test_endpoints_all_random() + # test.test_endpoints_on_off_continuous() + # test.test_endpoints_on_off_random() + # test.test_endpoints_specific_random() + test.generate_summary() + except KeyboardInterrupt: + print("Interrupted by user, shutting down...") + except Exception as e: + traceback.print_exc() + print(f"An error occurred: {e}") \ No newline at end of file diff --git a/src-python/controller.py b/src-python/controller.py index b10d5617..b31a2759 100644 --- a/src-python/controller.py +++ b/src-python/controller.py @@ -795,24 +795,28 @@ class Controller: @staticmethod def setEnableTranslation(*args, **kwargs) -> dict: - if model.isLoadedCTranslate2Model() is False: - model.changeTranslatorCTranslate2Model() - config.ENABLE_TRANSLATION = True + if config.ENABLE_TRANSLATION is False: + if model.isLoadedCTranslate2Model() is False: + model.changeTranslatorCTranslate2Model() + config.ENABLE_TRANSLATION = True return {"status":200, "result":config.ENABLE_TRANSLATION} @staticmethod def setDisableTranslation(*args, **kwargs) -> dict: - config.ENABLE_TRANSLATION = False + if config.ENABLE_TRANSLATION is True: + config.ENABLE_TRANSLATION = False return {"status":200, "result":config.ENABLE_TRANSLATION} @staticmethod def setEnableForeground(*args, **kwargs) -> dict: - config.ENABLE_FOREGROUND = True + if config.ENABLE_FOREGROUND is False: + config.ENABLE_FOREGROUND = True return {"status":200, "result":config.ENABLE_FOREGROUND} @staticmethod def setDisableForeground(*args, **kwargs) -> dict: - config.ENABLE_FOREGROUND = False + if config.ENABLE_FOREGROUND is True: + config.ENABLE_FOREGROUND = False return {"status":200, "result":config.ENABLE_FOREGROUND} @staticmethod @@ -906,12 +910,14 @@ class Controller: @staticmethod def setEnableConvertMessageToRomaji(*args, **kwargs) -> dict: - config.CONVERT_MESSAGE_TO_ROMAJI = True + 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: - config.CONVERT_MESSAGE_TO_ROMAJI = False + if config.CONVERT_MESSAGE_TO_ROMAJI is True: + config.CONVERT_MESSAGE_TO_ROMAJI = False return {"status":200, "result":config.CONVERT_MESSAGE_TO_ROMAJI} @staticmethod @@ -920,12 +926,14 @@ class Controller: @staticmethod def setEnableConvertMessageToHiragana(*args, **kwargs) -> dict: - config.CONVERT_MESSAGE_TO_HIRAGANA = True + 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: - config.CONVERT_MESSAGE_TO_HIRAGANA = False + if config.CONVERT_MESSAGE_TO_HIRAGANA is True: + config.CONVERT_MESSAGE_TO_HIRAGANA = False return {"status":200, "result":config.CONVERT_MESSAGE_TO_HIRAGANA} @staticmethod @@ -934,12 +942,14 @@ class Controller: @staticmethod def setEnableMainWindowSidebarCompactMode(*args, **kwargs) -> dict: - config.MAIN_WINDOW_SIDEBAR_COMPACT_MODE = True + 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: - config.MAIN_WINDOW_SIDEBAR_COMPACT_MODE = False + 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} @staticmethod @@ -993,12 +1003,14 @@ class Controller: @staticmethod def setEnableShowResendButton(*args, **kwargs) -> dict: - config.SHOW_RESEND_BUTTON = True + 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: - config.SHOW_RESEND_BUTTON = False + if config.SHOW_RESEND_BUTTON is True: + config.SHOW_RESEND_BUTTON = False return {"status":200, "result":config.SHOW_RESEND_BUTTON} @staticmethod @@ -1033,19 +1045,21 @@ class Controller: return {"status":200, "result":config.AUTO_MIC_SELECT} def setEnableAutoMicSelect(self, *args, **kwargs) -> dict: - config.AUTO_MIC_SELECT = True - device_manager.setCallbackProcessBeforeUpdateDevices(self.stopAccessDevices) - device_manager.setCallbackDefaultMicDevice(self.updateSelectedMicDevice) - device_manager.setCallbackProcessAfterUpdateDevices(self.restartAccessDevices) - device_manager.forceUpdateAndSetMicDevices() + 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: - device_manager.clearCallbackProcessBeforeUpdateDevices() - device_manager.clearCallbackDefaultMicDevice() - device_manager.clearCallbackProcessAfterUpdateDevices() - config.AUTO_MIC_SELECT = False + if config.AUTO_MIC_SELECT is True: + device_manager.clearCallbackProcessBeforeUpdateDevices() + device_manager.clearCallbackDefaultMicDevice() + device_manager.clearCallbackProcessAfterUpdateDevices() + config.AUTO_MIC_SELECT = False return {"status":200, "result":config.AUTO_MIC_SELECT} @staticmethod @@ -1108,12 +1122,14 @@ class Controller: @staticmethod def setEnableMicAutomaticThreshold(*args, **kwargs) -> dict: - config.MIC_AUTOMATIC_THRESHOLD = True + 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: - config.MIC_AUTOMATIC_THRESHOLD = False + if config.MIC_AUTOMATIC_THRESHOLD is True: + config.MIC_AUTOMATIC_THRESHOLD = False return {"status":200, "result":config.MIC_AUTOMATIC_THRESHOLD} @staticmethod @@ -1226,20 +1242,21 @@ class Controller: return {"status":200, "result":config.AUTO_SPEAKER_SELECT} def setEnableAutoSpeakerSelect(self, *args, **kwargs) -> dict: - config.AUTO_SPEAKER_SELECT = True - device_manager.setCallbackProcessBeforeUpdateDevices(self.stopAccessDevices) - device_manager.setCallbackDefaultSpeakerDevice(self.updateSelectedSpeakerDevice) - device_manager.setCallbackProcessAfterUpdateDevices(self.restartAccessDevices) - device_manager.forceUpdateAndSetSpeakerDevices() - + 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: - device_manager.clearCallbackProcessBeforeUpdateDevices() - device_manager.clearCallbackDefaultSpeakerDevice() - device_manager.clearCallbackProcessAfterUpdateDevices() - config.AUTO_SPEAKER_SELECT = False + if config.AUTO_SPEAKER_SELECT is True: + device_manager.clearCallbackProcessBeforeUpdateDevices() + device_manager.clearCallbackDefaultSpeakerDevice() + device_manager.clearCallbackProcessAfterUpdateDevices() + config.AUTO_SPEAKER_SELECT = False return {"status":200, "result":config.AUTO_SPEAKER_SELECT} @staticmethod @@ -1285,12 +1302,14 @@ class Controller: @staticmethod def setEnableSpeakerAutomaticThreshold(*args, **kwargs) -> dict: - config.SPEAKER_AUTOMATIC_THRESHOLD = True + 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: - config.SPEAKER_AUTOMATIC_THRESHOLD = False + if config.SPEAKER_AUTOMATIC_THRESHOLD is True: + config.SPEAKER_AUTOMATIC_THRESHOLD = False return {"status":200, "result":config.SPEAKER_AUTOMATIC_THRESHOLD} @staticmethod @@ -1459,12 +1478,14 @@ class Controller: @staticmethod def setEnableNotificationVrcSfx(*args, **kwargs) -> dict: - config.NOTIFICATION_VRC_SFX = True + 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: - config.NOTIFICATION_VRC_SFX = False + if config.NOTIFICATION_VRC_SFX is True: + config.NOTIFICATION_VRC_SFX = False return {"status":200, "result":config.NOTIFICATION_VRC_SFX} @staticmethod @@ -1571,12 +1592,14 @@ class Controller: @staticmethod def setEnableAutoClearMessageBox(*args, **kwargs) -> dict: - config.AUTO_CLEAR_MESSAGE_BOX = True + 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: - config.AUTO_CLEAR_MESSAGE_BOX = False + if config.AUTO_CLEAR_MESSAGE_BOX is True: + config.AUTO_CLEAR_MESSAGE_BOX = False return {"status":200, "result":config.AUTO_CLEAR_MESSAGE_BOX} @staticmethod @@ -1585,12 +1608,14 @@ class Controller: @staticmethod def setEnableSendOnlyTranslatedMessages(*args, **kwargs) -> dict: - config.SEND_ONLY_TRANSLATED_MESSAGES = True + 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: - config.SEND_ONLY_TRANSLATED_MESSAGES = False + if config.SEND_ONLY_TRANSLATED_MESSAGES is True: + config.SEND_ONLY_TRANSLATED_MESSAGES = False return {"status":200, "result":config.SEND_ONLY_TRANSLATED_MESSAGES} @staticmethod @@ -1599,17 +1624,19 @@ class Controller: @staticmethod def setEnableOverlaySmallLog(*args, **kwargs) -> dict: - config.OVERLAY_SMALL_LOG = True - if config.OVERLAY_LARGE_LOG is False: - model.startOverlay() + 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 - model.clearOverlayImageSmallLog() - if config.OVERLAY_LARGE_LOG is False: - model.shutdownOverlay() + 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_SMALL_LOG is False: - model.startOverlay() + 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 - model.clearOverlayImageLargeLog() - if config.OVERLAY_SMALL_LOG is False: - model.shutdownOverlay() + 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,12 +1686,14 @@ class Controller: @staticmethod def setEnableOverlayShowOnlyTranslatedMessages(*args, **kwargs) -> dict: - config.OVERLAY_SHOW_ONLY_TRANSLATED_MESSAGES = True + 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: - config.OVERLAY_SHOW_ONLY_TRANSLATED_MESSAGES = False + 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} @staticmethod @@ -1671,12 +1702,14 @@ class Controller: @staticmethod def setEnableSendMessageToVrc(*args, **kwargs) -> dict: - config.SEND_MESSAGE_TO_VRC = True + 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: - config.SEND_MESSAGE_TO_VRC = False + if config.SEND_MESSAGE_TO_VRC is True: + config.SEND_MESSAGE_TO_VRC = False return {"status":200, "result":config.SEND_MESSAGE_TO_VRC} @staticmethod @@ -1685,12 +1718,14 @@ class Controller: @staticmethod def setEnableSendReceivedMessageToVrc(*args, **kwargs) -> dict: - config.SEND_RECEIVED_MESSAGE_TO_VRC = True + 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: - config.SEND_RECEIVED_MESSAGE_TO_VRC = False + 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} @staticmethod @@ -1699,14 +1734,16 @@ class Controller: @staticmethod def setEnableLoggerFeature(*args, **kwargs) -> dict: - config.LOGGER_FEATURE = True - model.startLogger() + if config.LOGGER_FEATURE is False: + model.startLogger() + config.LOGGER_FEATURE = True return {"status":200, "result":config.LOGGER_FEATURE} @staticmethod def setDisableLoggerFeature(*args, **kwargs) -> dict: - model.stopLogger() - config.LOGGER_FEATURE = False + if config.LOGGER_FEATURE is True: + model.stopLogger() + config.LOGGER_FEATURE = False return {"status":200, "result":config.LOGGER_FEATURE} @staticmethod @@ -1715,45 +1752,53 @@ class Controller: @staticmethod def setEnableVrcMicMuteSync(*args, **kwargs) -> dict: - if model.getIsOscQueryEnabled() is True: - config.VRC_MIC_MUTE_SYNC = True - model.setMuteSelfStatus() - model.changeMicTranscriptStatus() - response = {"status":200, "result":config.VRC_MIC_MUTE_SYNC} + if config.VRC_MIC_MUTE_SYNC is False: + if model.getIsOscQueryEnabled() is True: + config.VRC_MIC_MUTE_SYNC = True + model.setMuteSelfStatus() + model.changeMicTranscriptStatus() + response = {"status":200, "result":config.VRC_MIC_MUTE_SYNC} + else: + response = { + "status":400, + "result":{ + "message":"Cannot enable VRC mic mute sync while OSC query is disabled", + "data": config.VRC_MIC_MUTE_SYNC + } + } else: - response = { - "status":400, - "result":{ - "message":"Cannot enable VRC mic mute sync while OSC query is disabled", - "data": config.VRC_MIC_MUTE_SYNC - } - } + response = {"status":200, "result":config.VRC_MIC_MUTE_SYNC} return response @staticmethod def setDisableVrcMicMuteSync(*args, **kwargs) -> dict: - config.VRC_MIC_MUTE_SYNC = False - model.changeMicTranscriptStatus() + 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: - self.startThreadingCheckSpeakerEnergy() - config.ENABLE_CHECK_ENERGY_RECEIVE = True + 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: - self.stopThreadingCheckSpeakerEnergy() - config.ENABLE_CHECK_ENERGY_RECEIVE = False + 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: - self.startThreadingCheckMicEnergy() - config.ENABLE_CHECK_ENERGY_SEND = True + 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: - self.stopThreadingCheckMicEnergy() - config.ENABLE_CHECK_ENERGY_SEND = False + 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} @staticmethod @@ -1767,23 +1812,27 @@ class Controller: return {"status":200, "result":True} def setEnableTranscriptionSend(self, *args, **kwargs) -> dict: - self.startThreadingTranscriptionSendMessage() - config.ENABLE_TRANSCRIPTION_SEND = True + 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: - self.stopThreadingTranscriptionSendMessage() - config.ENABLE_TRANSCRIPTION_SEND = False + 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: - self.startThreadingTranscriptionReceiveMessage() - config.ENABLE_TRANSCRIPTION_RECEIVE = True + 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: - self.stopThreadingTranscriptionReceiveMessage() - config.ENABLE_TRANSCRIPTION_RECEIVE = False + if config.ENABLE_TRANSCRIPTION_RECEIVE is True: + self.stopThreadingTranscriptionReceiveMessage() + config.ENABLE_TRANSCRIPTION_RECEIVE = False return {"status":200, "result":config.ENABLE_TRANSCRIPTION_RECEIVE} def sendMessageBox(self, data, *args, **kwargs) -> dict: @@ -2245,24 +2294,28 @@ class Controller: @staticmethod def setEnableWebSocketServer(*args, **kwargs) -> dict: - if isAvailableWebSocketServer(config.WEBSOCKET_HOST, config.WEBSOCKET_PORT) is True: - model.startWebSocketServer(config.WEBSOCKET_HOST, config.WEBSOCKET_PORT) - config.WEBSOCKET_SERVER = True - response = {"status":200, "result":config.WEBSOCKET_SERVER} - else: - response = { - "status":400, - "result":{ - "message":"WebSocket server host or port is not available", - "data": config.WEBSOCKET_SERVER + if config.WEBSOCKET_SERVER is False: + if isAvailableWebSocketServer(config.WEBSOCKET_HOST, config.WEBSOCKET_PORT) is True: + model.startWebSocketServer(config.WEBSOCKET_HOST, config.WEBSOCKET_PORT) + config.WEBSOCKET_SERVER = True + response = {"status":200, "result":config.WEBSOCKET_SERVER} + else: + response = { + "status":400, + "result":{ + "message":"WebSocket server host or port is not available", + "data": config.WEBSOCKET_SERVER + } } - } + else: + response = {"status":200, "result":config.WEBSOCKET_SERVER} return response @staticmethod def setDisableWebSocketServer(*args, **kwargs) -> dict: - config.WEBSOCKET_SERVER = False - model.stopWebSocketServer() + if config.WEBSOCKET_SERVER is True: + config.WEBSOCKET_SERVER = False + model.stopWebSocketServer() return {"status":200, "result":config.WEBSOCKET_SERVER} def initializationProgress(self, progress): diff --git a/src-python/mainloop.py b/src-python/mainloop.py index 0010b98a..6e5f6548 100644 --- a/src-python/mainloop.py +++ b/src-python/mainloop.py @@ -347,9 +347,11 @@ init_mapping = {key:value for key, value in mapping.items() if key.startswith("/ controller.setInitMapping(init_mapping) class Main: - def __init__(self) -> None: + def __init__(self, controller_instance, mapping_data) -> None: self.queue = Queue() self.main_loop = True + self.controller = controller_instance + self.mapping = mapping_data def receiver(self) -> None: while True: @@ -360,7 +362,7 @@ class Main: endpoint = received_data.get("endpoint", None) data = received_data.get("data", None) data = encodeBase64(data) if data is not None else None - printLog(endpoint, {"receive_data":data}) + printLog(endpoint, {"receive_data": data}) self.queue.put((endpoint, data)) def startReceiver(self) -> None: @@ -369,7 +371,10 @@ class Main: th_receiver.start() def handleRequest(self, endpoint, data=None) -> tuple: - handler = mapping.get(endpoint) + result = None # デフォルト値を設定 + status = 500 # デフォルト値を設定 + + handler = self.mapping.get(endpoint) if handler is None: response = "Invalid endpoint" status = 404 @@ -381,10 +386,12 @@ class Main: response = handler["variable"](data) status = response.get("status", None) result = response.get("result", None) + time.sleep(0.2) # 処理の安定化のために少し待機 except Exception as e: errorLogging() result = str(e) status = 500 + return result, status def handler(self) -> None: @@ -417,13 +424,15 @@ class Main: def stop(self) -> None: self.main_loop = False -if __name__ == "__main__": - main = Main() - main.startReceiver() - main.startHandler() +# 外部から参照可能なインスタンスを提供 +main_instance = Main(controller_instance=controller, mapping_data=mapping) - controller.setWatchdogCallback(main.stop) - controller.init() +if __name__ == "__main__": + main_instance.startReceiver() + main_instance.startHandler() + + main_instance.controller.setWatchdogCallback(main_instance.stop) + main_instance.controller.init() # mappingのすべてのstatusをTrueにする for key in mapping.keys(): @@ -432,13 +441,13 @@ if __name__ == "__main__": process = "main" match process: case "main": - main.start() + main_instance.start() case "test": for _ in range(100): time.sleep(0.5) endpoint = "/get/data/mic_host_list" - result, status = main.handleRequest(endpoint) + result, status = main_instance.handleRequest(endpoint) printResponse(status, endpoint, result) case "test_all": @@ -639,6 +648,6 @@ if __name__ == "__main__": case _: data = None - result, status = main.handleRequest(endpoint, data) + result, status = main_instance.handleRequest(endpoint, data) printResponse(status, endpoint, result) time.sleep(0.5) \ No newline at end of file