Add app configuration management

- Integrate Pydantic and Pydantic Settings for config models and validation
- Add `config.yaml` for structured and readable configuration
- Centralize environment variable loading and configuration instance

software/config.yaml

@@ -0,0 +1,33 @@
+client:
+  enabled: false
+  url: null
+  platform: auto
+
+llm:
+  service: litellm
+  model: gpt-4
+  vision_enabled: false
+  functions_enabled: false
+  context_window: 2048
+  max_tokens: 4096
+  temperature: 0.8
+
+local:
+  enabled: true
+  tts_service: piper
+  stt_service: local-whisper
+
+server:
+  enabled: false
+  host: 0.0.0.0
+  port: 10001
+
+stt:
+  service: openai
+
+tts:
+  service: openai
+
+tunnel:
+  service: ngrok
+  exposed: false

software/source/__init__.py

@@ -0,0 +1,6 @@
+from dotenv import load_dotenv
+from source.core.models import Config
+
+load_dotenv()
+
+config = Config()

software/source/core/models.py

@@ -0,0 +1,117 @@
+"""
+Application configuration models.
+"""
+
+from pydantic import BaseModel
+from pydantic_settings import (
+    BaseSettings,
+    PydanticBaseSettingsSource,
+    SettingsConfigDict,
+    YamlConfigSettingsSource,
+)
+
+APP_PREFIX = "01_"
+
+
+class Client(BaseModel):
+    """
+    Client configuration model
+    """
+
+    enabled: bool = False
+    url: None | str = None
+    platform: str = "auto"
+
+
+class LLM(BaseModel):
+    """
+    LLM configuration model
+    """
+
+    service: str = "litellm"
+    model: str = "gpt-4"
+    vision_enabled: bool = False
+    functions_enabled: bool = False
+    context_window: int = 2048
+    max_tokens: int = 4096
+    temperature: float = 0.8
+
+
+class Local(BaseModel):
+    """
+    Local configuration model
+    """
+
+    enabled: bool = False
+    tts_service: str = "piper"
+    stt_service: str = "local-whisper"
+
+
+class Server(BaseModel):
+    """
+    Server configuration model
+    """
+
+    enabled: bool = False
+    host: str = "0.0.0.0"
+    port: int = 10001
+
+
+class STT(BaseModel):
+    """
+    Speech-to-text configuration model
+    """
+
+    service: str = "openai"
+
+
+class TTS(BaseModel):
+    """
+    Text-to-speech configuration model
+    """
+
+    service: str = "openai"
+
+
+class Tunnel(BaseModel):
+    """
+    Tunnel configuration model
+    """
+
+    service: str = "ngrok"
+    exposed: bool = False
+
+
+class Config(BaseSettings):
+    """
+    Base configuration model
+    """
+
+    client: Client = Client()
+    llm: LLM = LLM()
+    local: Local = Local()
+    server: Server = Server()
+    stt: STT = STT()
+    tts: TTS = TTS()
+    tunnel: Tunnel = Tunnel()
+
+    model_config = SettingsConfigDict(
+        env_prefix=APP_PREFIX,
+        env_file=".env",
+        env_file_encoding="utf-8",
+        yaml_file="config.yaml",
+    )
+
+    @classmethod
+    def settings_customise_sources(
+        cls,
+        settings_cls: type[BaseSettings],
+        init_settings: PydanticBaseSettingsSource,
+        env_settings: PydanticBaseSettingsSource,
+        dotenv_settings: PydanticBaseSettingsSource,
+        file_secret_settings: PydanticBaseSettingsSource,
+    ) -> tuple[PydanticBaseSettingsSource, ...]:
+        """
+        Modify the order of precedence for settings sources.
+        """
+        return (YamlConfigSettingsSource(settings_cls),)

software/source/core/utils.py

@@ -0,0 +1,99 @@
+"""
+Core utilty functions for the server and client
+"""
+
+import asyncio
+import importlib
+import os
+import platform
+from threading import Thread
+from typing import NoReturn
+
+from source.server.server import main
+from source.server.tunnel import create_tunnel
+
+
+def get_client_platform(config) -> None:
+    """
+    Returns the client platform based on the system type.
+    """
+    if config.client.platform == "auto":
+        system_type: str = platform.system()
+
+        # macOS
+        if system_type == "Darwin":
+            config.client.platform = "mac"
+
+        # Linux
+        elif system_type == "Linux":
+            try:
+                with open("/proc/device-tree/model", "r", encoding="utf-8") as m:
+                    if "raspberry pi" in m.read().lower():
+                        config.client.platform = "rpi"
+                    else:
+                        config.client.platform = "linux"
+            except FileNotFoundError:
+                config.client.platform = "linux"
+
+
+def handle_exit(signum, frame) -> NoReturn:  # pylint: disable=unused-argument
+    """
+    Handle exit signal.
+    """
+    os._exit(0)
+
+
+def start_client(config) -> Thread:
+    """
+    Start the client.
+    """
+    module = importlib.import_module(
+        f".clients.{config.client.platform}.device", package="source"
+    )
+
+    client_thread = Thread(target=module.main, args=[config.client.url])
+    client_thread.start()
+    return client_thread
+
+
+def start_server(config) -> Thread:
+    """
+    Start the server.
+    """
+    loop: asyncio.AbstractEventLoop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+
+    server_thread = Thread(
+        target=loop.run_until_complete,
+        args=(
+            main(
+                config.server.host,
+                config.server.port,
+                config.llm.service,
+                config.llm.model,
+                config.llm.vision_enabled,
+                config.llm.functions_enabled,
+                config.llm.context_window,
+                config.llm.max_tokens,
+                config.llm.temperature,
+                config.tts.service,
+                config.stt.service,
+            ),
+        ),
+    )
+
+    server_thread.start()
+    return server_thread
+
+
+def start_tunnel(config) -> Thread:
+    """
+    Start the tunnel.
+    """
+    tunnel_thread = Thread(
+        target=create_tunnel,
+        args=[config.tunnel.service, config.server.host, config.server.port],
+    )
+
+    tunnel_thread.start()
+    return tunnel_thread

software/start.py

@@ -1,148 +1,68 @@
-import typer
+"""
-import asyncio
+Application entry point
-import platform
+"""
-import concurrent.futures
-import threading
-import os
-import importlib
-from source.server.tunnel import create_tunnel
-from source.server.server import main
-from source.server.utils.local_mode import select_local_model
 
 import signal
-app = typer.Typer()
+from threading import Thread
-
+
-@app.command()
+from source import config
-def run(
+from source.core.utils import (
-            server: bool = typer.Option(False, "--server", help="Run server"),
+    get_client_platform,
-            server_host: str = typer.Option("0.0.0.0", "--server-host", help="Specify the server host where the server will deploy"),
+    handle_exit,
-            server_port: int = typer.Option(10001, "--server-port", help="Specify the server port where the server will deploy"),
+    start_client,
-            
+    start_server,
-            tunnel_service: str = typer.Option("ngrok", "--tunnel-service", help="Specify the tunnel service"),
+    start_tunnel,
-            expose: bool = typer.Option(False, "--expose", help="Expose server to internet"),
-            
-            client: bool = typer.Option(False, "--client", help="Run client"),
-            server_url: str = typer.Option(None, "--server-url", help="Specify the server URL that the client should expect. Defaults to server-host and server-port"),
-            client_type: str = typer.Option("auto", "--client-type", help="Specify the client type"),
-            
-            llm_service: str = typer.Option("litellm", "--llm-service", help="Specify the LLM service"),
-            
-            model: str = typer.Option("gpt-4", "--model", help="Specify the model"),
-            llm_supports_vision: bool = typer.Option(False, "--llm-supports-vision", help="Specify if the LLM service supports vision"),
-            llm_supports_functions: bool = typer.Option(False, "--llm-supports-functions", help="Specify if the LLM service supports functions"),
-            context_window: int = typer.Option(2048, "--context-window", help="Specify the context window size"),
-            max_tokens: int = typer.Option(4096, "--max-tokens", help="Specify the maximum number of tokens"),
-            temperature: float = typer.Option(0.8, "--temperature", help="Specify the temperature for generation"),
-            
-            tts_service: str = typer.Option("openai", "--tts-service", help="Specify the TTS service"),
-            
-            stt_service: str = typer.Option("openai", "--stt-service", help="Specify the STT service"),
-
-            local: bool = typer.Option(False, "--local", help="Use recommended local services for LLM, STT, and TTS"),
-        ):
-    
-    _run(
-        server=server,
-        server_host=server_host,
-        server_port=server_port,
-        tunnel_service=tunnel_service,
-        expose=expose,
-        client=client,
-        server_url=server_url,
-        client_type=client_type,
-        llm_service=llm_service,
-        model=model,
-        llm_supports_vision=llm_supports_vision,
-        llm_supports_functions=llm_supports_functions,
-        context_window=context_window,
-        max_tokens=max_tokens,
-        temperature=temperature,
-        tts_service=tts_service,
-        stt_service=stt_service,
-        local=local
 )
+from source.server.utils.local_mode import select_local_model
 
-def _run(
-            server: bool = False,
-            server_host: str = "0.0.0.0",
-            server_port: int = 10001,
-            
-            tunnel_service: str = "bore",
-            expose: bool = False,
-            
-            client: bool = False,
-            server_url: str = None,
-            client_type: str = "auto",
-            
-            llm_service: str = "litellm",
-            
-            model: str = "gpt-4",
-            llm_supports_vision: bool = False,
-            llm_supports_functions: bool = False,
-            context_window: int = 2048,
-            max_tokens: int = 4096,
-            temperature: float = 0.8,
-            
-            tts_service: str = "openai",
 
-            stt_service: str = "openai",
+def run() -> None:
+    """
+    Run the application.
+    """
+    # Set up signal handler for SIGINT (keyboard interrupt)
+    signal.signal(signal.SIGINT, handle_exit)
 
-            local: bool = False
+    # If platform is set to auto, determine user's platform automatically.
-        ):
+    if config.client.platform == "auto":
+        get_client_platform(config)
 
-    if local:
+    # If local mode is enabled, set up local services
-        tts_service = "piper"
+    if config.local.enabled:
-        # llm_service = "llamafile"
+        config.tts.service = config.local.tts_service
-        stt_service = "local-whisper"
+        config.stt.service = config.local.stt_service
         select_local_model()
 
-    if not server_url:
+    # If no client URL is provided, set one using server host and port.
-        server_url = f"{server_host}:{server_port}"
+    config.client.url = (
-    
+        config.client.url or f"{config.server.host}:{config.server.port}"
-    if not server and not client:
+    )
-        server = True
-        client = True
-
-    def handle_exit(signum, frame):
-        os._exit(0)
-
-    signal.signal(signal.SIGINT, handle_exit)
 
-    if server:
+    if not config.server.enabled and not config.client.enabled:
-        loop = asyncio.new_event_loop()
+        config.server.enabled = config.client.enabled = True
-        asyncio.set_event_loop(loop)
-        server_thread = threading.Thread(target=loop.run_until_complete, args=(main(server_host, server_port, llm_service, model, llm_supports_vision, llm_supports_functions, context_window, max_tokens, temperature, tts_service, stt_service),))
-        server_thread.start()
 
-    if expose:
+    server_thread: Thread | None = (
-        tunnel_thread = threading.Thread(target=create_tunnel, args=[tunnel_service, server_host, server_port])
+        start_server(config) if config.server.enabled else None
-        tunnel_thread.start()
+    )
 
-    if client:
+    tunnel_thread: Thread | None = (
-        if client_type == "auto":
+        start_tunnel(config) if config.tunnel.exposed else None
-            system_type = platform.system()
+    )
-            if system_type == "Darwin":  # Mac OS
-                client_type = "mac"
-            elif system_type == "Linux":  # Linux System
-                try:
-                    with open('/proc/device-tree/model', 'r') as m:
-                        if 'raspberry pi' in m.read().lower():
-                            client_type = "rpi"
-                        else:
-                            client_type = "linux"
-                except FileNotFoundError:
-                    client_type = "linux"
 
-        module = importlib.import_module(f".clients.{client_type}.device", package='source')
+    client_thread: Thread | None = (
-        client_thread = threading.Thread(target=module.main, args=[server_url])
+        start_client(config) if config.client.enabled else None
-        client_thread.start()
+    )
 
     try:
-        if server:
+        if server_thread:
             server_thread.join()
-        if expose:
+        if tunnel_thread:
             tunnel_thread.join()
-        if client:
+        if client_thread and client_thread.is_alive():
             client_thread.join()
     except KeyboardInterrupt:
-        os.kill(os.getpid(), signal.SIGINT)
+        handle_exit(signal.SIGINT, None)
+
+
+if __name__ == "__main__":
+    run()