diff --git a/src-python/backend_test.py b/src-python/backend_test.py index c43a622d..ffa043e2 100644 --- a/src-python/backend_test.py +++ b/src-python/backend_test.py @@ -180,17 +180,21 @@ class TestMainloop(): case "/set/data/selected_tab_no": data = random.choice(["1", "2", "3"]) case "/set/data/selected_translation_engines": + print("Fetching endpoint data for translation_engines...") + self.config_dict["translation_engines"], _ = self.main.handleRequest("/get/data/translation_engines", None) 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": + self.config_dict["selectable_language_list"], _ = self.main.handleRequest("/get/data/selectable_language_list", None) 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": + self.config_dict["selectable_language_list"], _ = self.main.handleRequest("/get/data/selectable_language_list", None) selectable_language_list = self.config_dict.get("selectable_language_list", None) data = {} for i in ["1", "2", "3"]: @@ -198,6 +202,7 @@ class TestMainloop(): 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": + self.config_dict["transcription_engines"], _ = self.main.handleRequest("/get/data/transcription_engines", None) transcription_engines = self.config_dict.get("transcription_engines", None) data = random.choice(transcription_engines) case "/set/data/transparency": @@ -222,11 +227,17 @@ class TestMainloop(): "height": random.randint(600, 1080) } case "/set/data/selected_translation_compute_device": - data = random.choice(self.config_dict["translation_compute_device_list"]) + self.config_dict["translation_compute_device_list"], _ = self.main.handleRequest("/get/data/translation_compute_device_list", None) + translation_compute_device_list = self.config_dict.get("translation_compute_device_list", None) + data = random.choice(translation_compute_device_list) case "/set/data/selected_transcription_compute_device": - data = random.choice(self.config_dict["transcription_compute_device_list"]) + self.config_dict["transcription_compute_device_list"], _ = self.main.handleRequest("/get/data/transcription_compute_device_list", None) + transcription_compute_device_list = self.config_dict.get("transcription_compute_device_list", None) + data = random.choice(transcription_compute_device_list) case "/set/data/ctranslate2_weight_type": - data = random.choice(list(self.config_dict["selectable_ctranslate2_weight_type_dict"].keys())) + self.config_dict["selectable_ctranslate2_weight_type_dict"], _ = self.main.handleRequest("/get/data/selectable_ctranslate2_weight_type_dict", None) + selectable_ctranslate2_weight_type_dict = self.config_dict.get("selectable_ctranslate2_weight_type_dict", None) + data = random.choice(list(selectable_ctranslate2_weight_type_dict.keys())) # LLM / API Clients case "/set/data/plamo_model": # 事前にモデルリストを取得 @@ -279,9 +290,13 @@ class TestMainloop(): data = "OPENAI_DUMMY_KEY" expected_status = [200, 400] case "/set/data/selected_mic_host": - data = random.choice(self.config_dict["mic_host_list"]) + self.config_dict["mic_host_list"], _ = self.main.handleRequest("/get/data/mic_host_list", None) + mic_host_list = self.config_dict.get("mic_host_list", None) + data = random.choice(mic_host_list) case "/set/data/selected_mic_device": - data = random.choice(self.config_dict["mic_device_list"]) + self.config_dict["mic_device_list"], _ = self.main.handleRequest("/get/data/mic_device_list", None) + mic_device_list = self.config_dict.get("mic_device_list", None) + data = random.choice(mic_device_list) case "/set/data/mic_threshold": data = random.randint(-1000, 3000) if 0 <= data <= 2000: @@ -290,13 +305,17 @@ class TestMainloop(): expected_status = [400] case "/set/data/mic_record_timeout": data = random.randint(-1, 10) - if 0 <= data <= self.config_dict["mic_phrase_timeout"]: + self.config_dict["mic_phrase_timeout"], _ = self.main.handleRequest("/get/data/mic_phrase_timeout", None) + mic_phrase_timeout = self.config_dict.get("mic_phrase_timeout", None) + if 0 <= data <= 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: + self.config_dict["mic_record_timeout"], _ = self.main.handleRequest("/get/data/mic_record_timeout", None) + mic_record_timeout = self.config_dict.get("mic_record_timeout", None) + if mic_record_timeout <= data: pass else: expected_status = [400] @@ -329,7 +348,9 @@ class TestMainloop(): ] ) case "/set/data/selected_speaker_device": - data = random.choice(self.config_dict["speaker_device_list"]) + self.config_dict["speaker_device_list"], _ = self.main.handleRequest("/get/data/speaker_device_list", None) + speaker_device_list = self.config_dict.get("speaker_device_list", None) + data = random.choice(speaker_device_list) case "/set/data/speaker_threshold": data = random.randint(-1000, 5000) if 0 <= data <= 4000: @@ -338,13 +359,17 @@ class TestMainloop(): expected_status = [400] case "/set/data/speaker_record_timeout": data = random.randint(-1, 10) - if 0 <= data <= self.config_dict["speaker_phrase_timeout"]: + self.config_dict["speaker_phrase_timeout"], _ = self.main.handleRequest("/get/data/speaker_phrase_timeout", None) + speaker_phrase_timeout = self.config_dict.get("speaker_phrase_timeout", None) + if 0 <= data <= 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: + self.config_dict["speaker_record_timeout"], _ = self.main.handleRequest("/get/data/speaker_record_timeout", None) + speaker_record_timeout = self.config_dict.get("speaker_record_timeout", None) + if speaker_record_timeout <= data: pass else: expected_status = [400] @@ -359,7 +384,9 @@ class TestMainloop(): 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]) + self.config_dict["selectable_whisper_weight_type_dict"], _ = self.main.handleRequest("/get/data/selectable_whisper_weight_type_dict", None) + selectable_whisper_weight_type_dict = self.config_dict.get("selectable_whisper_weight_type_dict", None) + data = random.choice([key for key, value in selectable_whisper_weight_type_dict.items() if value is True]) case "/set/data/overlay_small_log_settings": data = { "x_pos": random.random(), @@ -389,9 +416,13 @@ class TestMainloop(): "tracker": random.choice(["HMD", "LeftHand", "RightHand"]), } case "/set/data/send_message_format_parts": - data = self.config_dict["send_message_format_parts"] + self.config_dict["send_message_format_parts"], _ = self.main.handleRequest("/get/data/send_message_format_parts", None) + send_message_format_parts = self.config_dict.get("send_message_format_parts", None) + data = send_message_format_parts case "/set/data/received_message_format_parts": - data = self.config_dict["received_message_format_parts"] + self.config_dict["received_message_format_parts"], _ = self.main.handleRequest("/get/data/received_message_format_parts", None) + received_message_format_parts = self.config_dict.get("received_message_format_parts", None) + data = 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": @@ -449,6 +480,8 @@ class TestMainloop(): print(f" Current config_dict: {self.config_dict}") else: print(f"-> {Color.YELLOW}[SKIP]{Color.RESET} No data to set for this endpoint: {endpoint}.") + status = None + result = None success = True self.record_test_result(endpoint, status, result if data is not None else None, expected_status) # テスト結果を記録 return success @@ -614,9 +647,9 @@ class TestMainloop(): ] self.set_data_specific_endpoints = [ - "/set/data/ctranslate2_weight_type", - "/set/data/websocket_host", - "/set/data/websocket_port", + # "/set/data/ctranslate2_weight_type", + # "/set/data/websocket_host", + # "/set/data/websocket_port", "/set/data/osc_ip_address", "/set/data/osc_port", ] @@ -625,7 +658,7 @@ class TestMainloop(): self.delete_data_endpoints = [] endpoint_types = [ - "validity", + # "validity", "set_data", # "run", # "delete", @@ -807,9 +840,9 @@ if __name__ == "__main__": # 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_continuous() # test.test_endpoints_on_off_random() - # test.test_endpoints_specific_random() + test.test_endpoints_specific_random() # test.test_translate_all_language_pairs() test.generate_summary() except KeyboardInterrupt: diff --git a/src-python/config.py b/src-python/config.py index 3444a4a3..24dd42e1 100644 --- a/src-python/config.py +++ b/src-python/config.py @@ -63,23 +63,201 @@ def _auto_register_descriptors(): make_serializer(name) +# Wrapper classes for mutable types that auto-save on modification +class ManagedDict(dict): + """Dict wrapper that saves changes back to config.""" + def __init__(self, instance, property_name, immediate_save): + self._instance = instance + self._property_name = property_name + self._immediate_save = immediate_save + self._internal_name = f"_{property_name}" + # Initialize from internal storage + super().__init__(getattr(instance, self._internal_name)) + + def _get_internal(self): + """Get reference to internal storage.""" + return getattr(self._instance, self._internal_name) + + def _save(self): + """Save current state back to config and sync internal storage.""" + try: + # Update internal storage directly + internal_dict = self._get_internal() + internal_dict.clear() + internal_dict.update(dict.items(self)) + # Trigger config save + self._instance.saveConfig(self._property_name, dict(self), immediate_save=self._immediate_save) + except Exception: + pass + + def __getitem__(self, key): + # Always read from internal storage to get latest value + return self._get_internal()[key] + + def __setitem__(self, key, value): + super().__setitem__(key, value) + self._save() + + def __delitem__(self, key): + super().__delitem__(key) + self._save() + + def __contains__(self, key): + return key in self._get_internal() + + def get(self, key, default=None): + return self._get_internal().get(key, default) + + def keys(self): + return self._get_internal().keys() + + def values(self): + return self._get_internal().values() + + def items(self): + return self._get_internal().items() + + def update(self, *args, **kwargs): + super().update(*args, **kwargs) + self._save() + + def pop(self, *args): + result = super().pop(*args) + self._save() + return result + + def popitem(self): + result = super().popitem() + self._save() + return result + + def clear(self): + super().clear() + self._save() + + def setdefault(self, key, default=None): + result = super().setdefault(key, default) + self._save() + return result + + +class ManagedList(list): + """List wrapper that saves changes back to config.""" + def __init__(self, instance, property_name, immediate_save): + self._instance = instance + self._property_name = property_name + self._immediate_save = immediate_save + self._internal_name = f"_{property_name}" + # Initialize from internal storage + super().__init__(getattr(instance, self._internal_name)) + + def _get_internal(self): + """Get reference to internal storage.""" + return getattr(self._instance, self._internal_name) + + def _save(self): + """Save current state back to config and sync internal storage.""" + try: + # Update internal storage directly + internal_list = self._get_internal() + internal_list.clear() + internal_list.extend(list.__iter__(self)) + # Trigger config save + self._instance.saveConfig(self._property_name, list(self), immediate_save=self._immediate_save) + except Exception: + pass + + def __getitem__(self, index): + # Always read from internal storage to get latest value + return self._get_internal()[index] + + def __setitem__(self, index, value): + super().__setitem__(index, value) + self._save() + + def __delitem__(self, index): + super().__delitem__(index) + self._save() + + def __len__(self): + return len(self._get_internal()) + + def __contains__(self, value): + return value in self._get_internal() + + def __iter__(self): + return iter(self._get_internal()) + + def append(self, value): + super().append(value) + self._save() + + def extend(self, iterable): + super().extend(iterable) + self._save() + + def insert(self, index, value): + super().insert(index, value) + self._save() + + def remove(self, value): + super().remove(value) + self._save() + + def pop(self, index=-1): + result = super().pop(index) + self._save() + return result + + def clear(self): + super().clear() + self._save() + + def sort(self, *args, **kwargs): + super().sort(*args, **kwargs) + self._save() + + def reverse(self): + super().reverse() + self._save() + + # Descriptor for simple managed config properties to reduce repetitive getters/setters. # It performs optional type validation, optional allowed-values check, and calls # instance.saveConfig(...) on successful set. class ManagedProperty: - def __init__(self, name: str, type_: type = None, allowed=None, immediate_save: bool = False, serialize: bool = True, readonly: bool = False): + def __init__(self, name: str, type_: type = None, allowed=None, immediate_save: bool = False, serialize: bool = True, readonly: bool = False, mutable_tracking: bool = False): self.name = name self.type_ = type_ self.allowed = allowed self.immediate_save = immediate_save self.serialize = serialize self.readonly = readonly + self.mutable_tracking = mutable_tracking self.private_name = f"_{name}" + self.wrapper_cache_name = f"_wrapper_{name}" def __get__(self, instance, owner): if instance is None: return self stored = getattr(instance, self.private_name) + + # If mutable_tracking is enabled, return cached wrapper or create new one + if self.mutable_tracking and isinstance(stored, dict): + wrapper = getattr(instance, self.wrapper_cache_name, None) + if wrapper is None or not isinstance(wrapper, ManagedDict): + wrapper = ManagedDict(instance, self.name, self.immediate_save) + setattr(instance, self.wrapper_cache_name, wrapper) + # Wrapper automatically syncs with internal storage on access + return wrapper + elif self.mutable_tracking and isinstance(stored, list): + wrapper = getattr(instance, self.wrapper_cache_name, None) + if wrapper is None or not isinstance(wrapper, ManagedList): + wrapper = ManagedList(instance, self.name, self.immediate_save) + setattr(instance, self.wrapper_cache_name, wrapper) + # Wrapper automatically syncs with internal storage on access + return wrapper + # Return deep copy for mutable types to prevent external modification if isinstance(stored, (dict, list)): return copy.deepcopy(stored) @@ -426,15 +604,15 @@ class Config: # --- Selectable dict/list properties (managed by descriptor, not serialized) --- # These are dynamically generated in init_config() based on installed packages/APIs - SELECTABLE_CTRANSLATE2_WEIGHT_TYPE_DICT = ManagedProperty('SELECTABLE_CTRANSLATE2_WEIGHT_TYPE_DICT', type_=dict, serialize=False) - SELECTABLE_WHISPER_WEIGHT_TYPE_DICT = ManagedProperty('SELECTABLE_WHISPER_WEIGHT_TYPE_DICT', type_=dict, serialize=False) - SELECTABLE_TRANSLATION_ENGINE_STATUS = ManagedProperty('SELECTABLE_TRANSLATION_ENGINE_STATUS', type_=dict, serialize=False) - SELECTABLE_TRANSCRIPTION_ENGINE_STATUS = ManagedProperty('SELECTABLE_TRANSCRIPTION_ENGINE_STATUS', type_=dict, serialize=False) - SELECTABLE_PLAMO_MODEL_LIST = ManagedProperty('SELECTABLE_PLAMO_MODEL_LIST', type_=list, serialize=False) - SELECTABLE_GEMINI_MODEL_LIST = ManagedProperty('SELECTABLE_GEMINI_MODEL_LIST', type_=list, serialize=False) - SELECTABLE_OPENAI_MODEL_LIST = ManagedProperty('SELECTABLE_OPENAI_MODEL_LIST', type_=list, serialize=False) - SELECTABLE_LMSTUDIO_MODEL_LIST = ManagedProperty('SELECTABLE_LMSTUDIO_MODEL_LIST', type_=list, serialize=False) - SELECTABLE_OLLAMA_MODEL_LIST = ManagedProperty('SELECTABLE_OLLAMA_MODEL_LIST', type_=list, serialize=False) + SELECTABLE_CTRANSLATE2_WEIGHT_TYPE_DICT = ManagedProperty('SELECTABLE_CTRANSLATE2_WEIGHT_TYPE_DICT', type_=dict, serialize=False, mutable_tracking=True) + SELECTABLE_WHISPER_WEIGHT_TYPE_DICT = ManagedProperty('SELECTABLE_WHISPER_WEIGHT_TYPE_DICT', type_=dict, serialize=False, mutable_tracking=True) + SELECTABLE_TRANSLATION_ENGINE_STATUS = ManagedProperty('SELECTABLE_TRANSLATION_ENGINE_STATUS', type_=dict, serialize=False, mutable_tracking=True) + SELECTABLE_TRANSCRIPTION_ENGINE_STATUS = ManagedProperty('SELECTABLE_TRANSCRIPTION_ENGINE_STATUS', type_=dict, serialize=False, mutable_tracking=True) + SELECTABLE_PLAMO_MODEL_LIST = ManagedProperty('SELECTABLE_PLAMO_MODEL_LIST', type_=list, serialize=False, mutable_tracking=True) + SELECTABLE_GEMINI_MODEL_LIST = ManagedProperty('SELECTABLE_GEMINI_MODEL_LIST', type_=list, serialize=False, mutable_tracking=True) + SELECTABLE_OPENAI_MODEL_LIST = ManagedProperty('SELECTABLE_OPENAI_MODEL_LIST', type_=list, serialize=False, mutable_tracking=True) + SELECTABLE_LMSTUDIO_MODEL_LIST = ManagedProperty('SELECTABLE_LMSTUDIO_MODEL_LIST', type_=list, serialize=False, mutable_tracking=True) + SELECTABLE_OLLAMA_MODEL_LIST = ManagedProperty('SELECTABLE_OLLAMA_MODEL_LIST', type_=list, serialize=False, mutable_tracking=True) # --- Save Json Data (ManagedProperty-based) --- # More simple boolean flags replaced with ManagedProperty diff --git a/src-python/controller.py b/src-python/controller.py index 60764865..02c42e6b 100644 --- a/src-python/controller.py +++ b/src-python/controller.py @@ -201,9 +201,7 @@ class Controller: def downloaded(self) -> None: 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 + config.SELECTABLE_CTRANSLATE2_WEIGHT_TYPE_DICT[self.weight_type] = True self.run( 200, @@ -236,9 +234,7 @@ class Controller: def downloaded(self) -> None: 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 + config.SELECTABLE_WHISPER_WEIGHT_TYPE_DICT[self.weight_type] = True self.run( 200, @@ -2649,10 +2645,8 @@ class Controller: return cleaned_text def updateDownloadedCTranslate2ModelWeight(self) -> None: - weight_type_dict = config.SELECTABLE_CTRANSLATE2_WEIGHT_TYPE_DICT - for weight_type in weight_type_dict.keys(): - weight_type_dict[weight_type] = model.checkTranslatorCTranslate2ModelWeight(weight_type) - config.SELECTABLE_CTRANSLATE2_WEIGHT_TYPE_DICT = weight_type_dict + for weight_type in config.SELECTABLE_CTRANSLATE2_WEIGHT_TYPE_DICT.keys(): + config.SELECTABLE_CTRANSLATE2_WEIGHT_TYPE_DICT[weight_type] = model.checkTranslatorCTranslate2ModelWeight(weight_type) def updateTranslationEngineAndEngineList(self): engines = config.SELECTED_TRANSLATION_ENGINES @@ -2674,10 +2668,8 @@ class Controller: self.run(200, self.run_mapping["translation_engines"], selectable_engines) def updateDownloadedWhisperModelWeight(self) -> None: - weight_type_dict = config.SELECTABLE_WHISPER_WEIGHT_TYPE_DICT - for weight_type in weight_type_dict.keys(): - weight_type_dict[weight_type] = model.checkTranscriptionWhisperModelWeight(weight_type) - config.SELECTABLE_WHISPER_WEIGHT_TYPE_DICT = weight_type_dict + for weight_type in config.SELECTABLE_WHISPER_WEIGHT_TYPE_DICT.keys(): + config.SELECTABLE_WHISPER_WEIGHT_TYPE_DICT[weight_type] = model.checkTranscriptionWhisperModelWeight(weight_type) def updateTranscriptionEngine(self): weight_type = config.WHISPER_WEIGHT_TYPE