From 521e60739b5e891ab42e45fcd6a324f0c8a3122f Mon Sep 17 00:00:00 2001 From: ekoc Date: Sun, 10 Nov 2024 21:50:04 +0100 Subject: [PATCH] Adding file change check and moving code to abstract class --- config.py | 102 +++++++++++++++++++++++++++++++------------- test/test_config.py | 19 +++++++++ 2 files changed, 92 insertions(+), 29 deletions(-) diff --git a/config.py b/config.py index 5ef892c..fc8dcf2 100644 --- a/config.py +++ b/config.py @@ -1,9 +1,13 @@ import os import json +import time import inspect from pathlib import Path from types import SimpleNamespace from dataclasses import dataclass +from abc import ABC, abstractmethod +from typing import Callable, Any + @dataclass class AlltalkConfigTheme: @@ -72,44 +76,76 @@ class AlltalkConfigDebug: debug_tts_variables = False debug_rvc = False -class AlltalkTTSEnginesConfig: +class AbstractConfig(ABC): + + def __init__(self, config_path: Path | str, file_check_interval: int): + self.__config_path = Path(config_path) if type(config_path) is str else config_path + self.__last_read_time = 0 # Track when we last read the file + self.__file_check_interval = file_check_interval + + def get_config_path(self): + return self.__config_path + + def reload(self): + self._load_config() + return self + + def _reload_on_change(self): + # Check if config file has been modified and reload if needed + if time.time() - self.__last_read_time >= self.__file_check_interval: + try: + most_recent_modification = self.get_config_path().stat().st_mtime + if most_recent_modification > self.__last_read_time: + 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 _object_hook(self) -> Callable[[dict[Any, Any]], Any] | None: + return None + + @abstractmethod + def _handle_loaded_config(self, data): + pass + +class AlltalkTTSEnginesConfig(AbstractConfig): __instance = None __this_dir = Path(__file__).parent.resolve() - def __init__(self): + 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.engine_loaded = "" self.selected_model = "" - + self._load_config() @staticmethod def get_instance(): if AlltalkTTSEnginesConfig.__instance is None: AlltalkTTSEnginesConfig.__instance = AlltalkTTSEnginesConfig() - AlltalkTTSEnginesConfig.__instance.__load_config() - return AlltalkTTSEnginesConfig.__instance - - def reload(self): - self.__load_config() - - def __load_config(self): - tts_engines_file = os.path.join(self.__this_dir, "system", "tts_engines", "tts_engines.json") - with open(tts_engines_file, "r") as f: - tts_engines_data = json.load(f) + return AlltalkTTSEnginesConfig.__instance._reload_on_change() + def _handle_loaded_config(self, data): # List of the available TTS engines from tts_engines.json - self.engines_available = [engine["name"] for engine in tts_engines_data["engines_available"]] + self.engines_available = [engine["name"] for engine in data["engines_available"]] # The currently set TTS engine from tts_engines.json - self.engine_loaded = tts_engines_data["engine_loaded"] - self.selected_model = tts_engines_data["selected_model"] + self.engine_loaded = data["engine_loaded"] + self.selected_model = data["selected_model"] - -class AlltalkConfig: +class AlltalkConfig(AbstractConfig): __instance = None __this_dir = Path(__file__).parent.resolve() - def __init__(self): + def __init__(self, config_path: Path | str = __this_dir / "confignew.json"): + super().__init__(config_path, 5) self.branding = "" self.delete_output_wavs = "" self.gradio_interface = False @@ -124,29 +160,37 @@ class AlltalkConfig: self.tgwui = AlltalkConfigTgwUi() self.api_def = AlltalkConfigApiDef() self.debugging = AlltalkConfigDebug() + self._load_config() @staticmethod def get_instance(): if AlltalkConfig.__instance is None: AlltalkConfig.__instance = AlltalkConfig() - AlltalkConfig.__instance.__load_config() - return AlltalkConfig.__instance + return AlltalkConfig.__instance._reload_on_change() - def reload(self): - self.__load_config() + def save(self, path: Path | None = None): + configfile_path = path if path is not None else self.get_config_path() - def __load_config(self): - configfile_path = self.__this_dir / "confignew.json" - with open(configfile_path, "r") as configfile: - configfile_data = json.load(configfile, object_hook=lambda d: SimpleNamespace(**d)) + # 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 _object_hook(self) -> Callable[[dict[Any, Any]], Any] | None: + return lambda d: SimpleNamespace(**d) + + def _handle_loaded_config(self, data): # Copy those fields for which there are members in this class: - for n, v in inspect.getmembers(configfile_data): + for n, v in inspect.getmembers(data): if hasattr(self, n) and not n.startswith("__"): setattr(self, n, v) # Special properties (cannot use 'class' as property name): - self.theme.clazz = configfile_data.theme.__dict__["class"] + self.theme.clazz = data.theme.__dict__["class"] # As a side effect, create the output directory output_directory = self.__this_dir / self.output_folder diff --git a/test/test_config.py b/test/test_config.py index 529f224..1f94f91 100644 --- a/test/test_config.py +++ b/test/test_config.py @@ -1,3 +1,5 @@ +import os.path +import tempfile import unittest from pathlib import Path @@ -10,6 +12,10 @@ class TestAlltalkConfig(unittest.TestCase): self.config = AlltalkConfig.get_instance() self.ttsEnginesConfig = AlltalkTTSEnginesConfig.get_instance() + def test_default_config_path(self): + expected_config_path = Path(__file__).parent.parent.resolve() / "confignew.json" + self.assertEqual(self.config.get_config_path(), expected_config_path) + def test_branding(self): self.assertEqual(self.config.branding, "AllTalk ") @@ -101,7 +107,20 @@ class TestAlltalkConfig(unittest.TestCase): self.assertFalse(self.config.debugging.debug_tts_variables) self.assertFalse(self.config.debugging.debug_rvc) + def test_save_config(self): + with tempfile.NamedTemporaryFile(suffix=".json") as tmp: + self.config.branding = "foo" + self.config.save(tmp.name) + new_config = AlltalkConfig(tmp.name) + self.assertEqual(new_config.branding, "foo") + 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"]) + + 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)) + +