James C. Palmer
2 years ago
parent
94edb8e001
commit
fe2a343c41
@ -1,6 +1,11 @@
|
||||
"""
|
||||
Loads environment variables and creates a global configuration object.
|
||||
"""
|
||||
|
||||
from dotenv import load_dotenv
|
||||
from source.core.models import Config
|
||||
|
||||
from source.core.config import Config, get_config
|
||||
|
||||
load_dotenv()
|
||||
|
||||
config = Config()
|
||||
config: Config = get_config()
|
||||
|
||||
@ -0,0 +1,110 @@
|
||||
"""
|
||||
Application configuration model.
|
||||
"""
|
||||
|
||||
import os
|
||||
from functools import lru_cache
|
||||
from typing import Any
|
||||
|
||||
from pydantic_settings import (
|
||||
BaseSettings,
|
||||
DotEnvSettingsSource,
|
||||
PydanticBaseSettingsSource,
|
||||
SettingsConfigDict,
|
||||
YamlConfigSettingsSource,
|
||||
)
|
||||
|
||||
from source.core.models import LLM, STT, TTS, Client, Local, Server, Tunnel
|
||||
|
||||
APP_PREFIX: str = os.getenv("01_PREFIX", "01_")
|
||||
|
||||
|
||||
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(extra="allow")
|
||||
|
||||
@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 (
|
||||
DotEnvSettingsSource(
|
||||
settings_cls,
|
||||
env_prefix=APP_PREFIX,
|
||||
env_file=".env",
|
||||
env_file_encoding="utf-8",
|
||||
env_nested_delimiter="_",
|
||||
),
|
||||
YamlConfigSettingsSource(
|
||||
settings_cls,
|
||||
yaml_file=os.getenv(f"{APP_PREFIX}CONFIG_FILE", "config.yaml"),
|
||||
),
|
||||
)
|
||||
|
||||
def apply_cli_args(self, args: dict) -> None:
|
||||
"""
|
||||
Apply CLI arguments to config.
|
||||
"""
|
||||
mapping: dict[str, str] = {
|
||||
"server": "server.enabled",
|
||||
"server_host": "server.host",
|
||||
"server_port": "server.port",
|
||||
"tunnel_service": "tunnel.service",
|
||||
"expose": "tunnel.exposed",
|
||||
"client": "client.enabled",
|
||||
"server_url": "client.url",
|
||||
"client_type": "client.platform",
|
||||
"llm_service": "llm.service",
|
||||
"model": "llm.model",
|
||||
"llm_supports_vision": "llm.vision_enabled",
|
||||
"llm_supports_functions": "llm.functions_enabled",
|
||||
"context_window": "llm.context_window",
|
||||
"max_tokens": "llm.max_tokens",
|
||||
"temperature": "llm.temperature",
|
||||
"tts_service": "tts.service",
|
||||
"stt_service": "stt.service",
|
||||
"local": "local.enabled",
|
||||
}
|
||||
|
||||
for key, path in mapping.items():
|
||||
if key in args and args[key] is not None:
|
||||
self.set_field(path, args[key])
|
||||
|
||||
def set_field(self, field: str, value: Any) -> None:
|
||||
"""
|
||||
Set field value
|
||||
"""
|
||||
obj: Any = self
|
||||
parts: list[str] = field.split(".")
|
||||
|
||||
for part in parts[:-1]:
|
||||
obj: Any = getattr(obj, part)
|
||||
|
||||
setattr(obj, parts[-1], value)
|
||||
|
||||
|
||||
@lru_cache()
|
||||
def get_config() -> Config:
|
||||
"""
|
||||
Return the application configuration.
|
||||
"""
|
||||
return Config()
|
||||
@ -1,99 +0,0 @@
|
||||
"""
|
||||
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
|
||||
@ -0,0 +1,12 @@
|
||||
"""
|
||||
System utility functions
|
||||
"""
|
||||
|
||||
import os
|
||||
|
||||
|
||||
def handle_exit(signum, frame) -> None: # pylint: disable=unused-argument
|
||||
"""
|
||||
Handle exit signal.
|
||||
"""
|
||||
os._exit(0)
|
||||
Loading…
Reference in new issue