Adding file change check and moving code to abstract class

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._reload_on_change()
-        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)
 
+    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.engine_loaded = data["engine_loaded"]
-        self.selected_model = tts_engines_data["selected_model"]
+        self.selected_model = data["selected_model"]
 
-
+class AlltalkConfig(AbstractConfig):
-class AlltalkConfig:
     __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._reload_on_change()
-        return AlltalkConfig.__instance
 
-    def reload(self):
+    def save(self, path: Path | None = None):
-        self.__load_config()
+        configfile_path = path if path is not None else self.get_config_path()
 
-    def __load_config(self):
+        # Remove private fields:
-        configfile_path = self.__this_dir / "confignew.json"
+        without_privates = {}
-        with open(configfile_path, "r") as configfile:
+        for attr, value in self.__dict__.items():
-            configfile_data = json.load(configfile, object_hook=lambda d: SimpleNamespace(**d))
+            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

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))
+
+