diff --git a/config.py b/config.py index fc8dcf2..d625895 100644 --- a/config.py +++ b/config.py @@ -2,11 +2,13 @@ import os import json import time import inspect +import shutil from pathlib import Path +from filelock import FileLock from types import SimpleNamespace from dataclasses import dataclass from abc import ABC, abstractmethod -from typing import Callable, Any +from typing import Callable, Any, MutableSequence @dataclass @@ -76,6 +78,12 @@ class AlltalkConfigDebug: debug_tts_variables = False debug_rvc = False +@dataclass +class AlltalkAvailableEngine: + name = "" + selected_model = "" + + class AbstractConfig(ABC): def __init__(self, config_path: Path | str, file_check_interval: int): @@ -99,47 +107,93 @@ class AbstractConfig(ABC): self.reload() except Exception as e: print(f"Error checking config file: {e}") - return self def _load_config(self): self.__last_read_time = self.get_config_path().stat().st_mtime - with open(self.get_config_path(), "r") as configfile: - data = json.load(configfile, object_hook=self._object_hook()) - - self._handle_loaded_config(data) + def __load(): + with open(self.get_config_path(), "r") as configfile: + data = json.load(configfile, object_hook=self._object_hook()) + self._handle_loaded_config(data) + self.__with_lock_and_backup(self.get_config_path(), False, __load) def _object_hook(self) -> Callable[[dict[Any, Any]], Any] | None: return None + def _save_file(self, path: Path | None | str, default = None, indent = 4): + file_path = (Path(path) if type(path) is str else path) if path is not None else self.get_config_path() + + # Remove private fields: + without_private_fields = {} + for attr, value in self.__dict__.items(): + if not attr.startswith("_"): + without_private_fields[attr] = value + + def __save(): + with open(file_path, "w") as file: + json.dump(without_private_fields, file, indent=indent, default=default) + self.__with_lock_and_backup(file_path, True, __save) + + def __with_lock_and_backup(self, path: Path, backup: bool, callable: Callable[[], None]): + lock_path = path.with_suffix('.lock') + try: + with FileLock(lock_path): + # Create backup: + if path.exists() and backup: + backup_path = path.with_suffix('.backup') + shutil.copy(path, backup_path) + + try: + callable() + except Exception as e: + if backup_path.exists(): + shutil.copy(backup_path, path) + raise Exception(f"Failed to save config: {e}") + finally: + # Cleanup lock and backup files: + lock_path.unlink() + if backup and backup_path.exists(): + backup_path.unlink() + + @abstractmethod def _handle_loaded_config(self, data): pass + class AlltalkTTSEnginesConfig(AbstractConfig): __instance = None __this_dir = Path(__file__).parent.resolve() def __init__(self, config_path: Path | str = os.path.join(__this_dir, "system", "tts_engines", "tts_engines.json")): super().__init__(config_path, 5) - self.engines_available = [] + self.engines_available: MutableSequence[AlltalkAvailableEngine] = [] self.engine_loaded = "" self.selected_model = "" self._load_config() + def get_engine_names_available(self): + return self.__engines_names_available + @staticmethod def get_instance(): if AlltalkTTSEnginesConfig.__instance is None: AlltalkTTSEnginesConfig.__instance = AlltalkTTSEnginesConfig() - return AlltalkTTSEnginesConfig.__instance._reload_on_change() + AlltalkTTSEnginesConfig.__instance._reload_on_change() + return AlltalkTTSEnginesConfig.__instance def _handle_loaded_config(self, data): # List of the available TTS engines from tts_engines.json - self.engines_available = [engine["name"] for engine in data["engines_available"]] + self.engines_available = data["engines_available"] + self.__engines_names_available = [engine["name"] for engine in data["engines_available"]] # The currently set TTS engine from tts_engines.json self.engine_loaded = data["engine_loaded"] self.selected_model = data["selected_model"] + def save(self, path: Path | str | None = None): + self._save_file(path) + + class AlltalkConfig(AbstractConfig): __instance = None __this_dir = Path(__file__).parent.resolve() @@ -166,19 +220,11 @@ class AlltalkConfig(AbstractConfig): def get_instance(): if AlltalkConfig.__instance is None: AlltalkConfig.__instance = AlltalkConfig() - return AlltalkConfig.__instance._reload_on_change() + AlltalkConfig.__instance._reload_on_change() + return AlltalkConfig.__instance - def save(self, path: Path | None = None): - configfile_path = path if path is not None else self.get_config_path() - - # Remove private fields: - without_privates = {} - for attr, value in self.__dict__.items(): - if not attr.startswith("_"): - without_privates[attr] = value - - with open(configfile_path, "w") as json_file: - json.dump(without_privates, json_file, indent=4, default=lambda o: o.__dict__) + def save(self, path: Path | str | None = None): + self._save_file(path, lambda o: o.__dict__) def _object_hook(self) -> Callable[[dict[Any, Any]], Any] | None: return lambda d: SimpleNamespace(**d) @@ -192,6 +238,6 @@ class AlltalkConfig(AbstractConfig): # Special properties (cannot use 'class' as property name): self.theme.clazz = data.theme.__dict__["class"] - # As a side effect, create the output directory + # As a side effect, create the output directory: output_directory = self.__this_dir / self.output_folder output_directory.mkdir(parents=True, exist_ok=True) diff --git a/system/requirements/requirements_colab.txt b/system/requirements/requirements_colab.txt index 6012944..baa415a 100644 --- a/system/requirements/requirements_colab.txt +++ b/system/requirements/requirements_colab.txt @@ -1,5 +1,6 @@ coqui-tts>=0.24.3 faster-whisper>=1.0.0 +filelock>=3.16.1 fuzzywuzzy gradio==4.32.2 importlib_metadata>=7.1.0 diff --git a/system/requirements/requirements_standalone.txt b/system/requirements/requirements_standalone.txt index e49cccc..7818dd3 100644 --- a/system/requirements/requirements_standalone.txt +++ b/system/requirements/requirements_standalone.txt @@ -1,4 +1,5 @@ coqui-tts>=0.24.3 +filelock>=3.16.1 fuzzywuzzy>=0.18.0 gradio>=4.44.1 importlib_metadata>=7.2.1 diff --git a/system/requirements/requirements_textgen.txt b/system/requirements/requirements_textgen.txt index dd874c0..84a38c9 100644 --- a/system/requirements/requirements_textgen.txt +++ b/system/requirements/requirements_textgen.txt @@ -1,4 +1,5 @@ coqui-tts>=0.24.3 +filelock>=3.16.1 fuzzywuzzy>=0.18.0 importlib_metadata>=7.1.0 inputimeout>=1.0.4 diff --git a/test/test_config.py b/test/test_config.py index 1f94f91..df1f4dd 100644 --- a/test/test_config.py +++ b/test/test_config.py @@ -117,10 +117,17 @@ class TestAlltalkConfig(unittest.TestCase): def test_tts_engines(self): self.assertEqual(self.ttsEnginesConfig.engine_loaded, "piper") self.assertEqual(self.ttsEnginesConfig.selected_model, "piper") - self.assertListEqual(self.ttsEnginesConfig.engines_available, ["parler", "piper", "vits", "xtts"]) + self.assertListEqual(self.ttsEnginesConfig.get_engine_names_available(), ["parler", "piper", "vits", "xtts"]) def test_tts_engines_default_config_path(self): expected_config_path = os.path.join(Path(__file__).parent.parent.resolve(), "system", "tts_engines", "tts_engines.json") self.assertEqual(self.ttsEnginesConfig.get_config_path(), Path(expected_config_path)) + def test_tts_engines_save(self): + with tempfile.NamedTemporaryFile(suffix=".json") as tmp: + self.ttsEnginesConfig.engine_loaded = "foo" + self.ttsEnginesConfig.save(tmp.name) + new_config = AlltalkTTSEnginesConfig(tmp.name) + self.assertEqual(new_config.engine_loaded, "foo") +