120 lines
3.8 KiB
Python
120 lines
3.8 KiB
Python
"""
|
|
Configuration Manager - Singleton pattern
|
|
|
|
Provides unified access to configuration with automatic validation.
|
|
"""
|
|
from pathlib import Path
|
|
from typing import Any, Optional
|
|
from loguru import logger
|
|
from .schema import ReelForgeConfig
|
|
from .loader import load_config_dict, save_config_dict
|
|
|
|
|
|
class ConfigManager:
|
|
"""
|
|
Configuration Manager (Singleton)
|
|
|
|
Provides unified access to configuration with automatic validation.
|
|
"""
|
|
_instance: Optional['ConfigManager'] = None
|
|
|
|
def __new__(cls, config_path: str = "config.yaml"):
|
|
if cls._instance is None:
|
|
cls._instance = super().__new__(cls)
|
|
return cls._instance
|
|
|
|
def __init__(self, config_path: str = "config.yaml"):
|
|
# Only initialize once
|
|
if hasattr(self, '_initialized'):
|
|
return
|
|
|
|
self.config_path = Path(config_path)
|
|
self.config: ReelForgeConfig = self._load()
|
|
self._initialized = True
|
|
|
|
def _load(self) -> ReelForgeConfig:
|
|
"""Load configuration from file"""
|
|
data = load_config_dict(str(self.config_path))
|
|
return ReelForgeConfig(**data)
|
|
|
|
def reload(self):
|
|
"""Reload configuration from file"""
|
|
self.config = self._load()
|
|
logger.info("Configuration reloaded")
|
|
|
|
def save(self):
|
|
"""Save current configuration to file"""
|
|
save_config_dict(self.config.to_dict(), str(self.config_path))
|
|
|
|
def update(self, updates: dict):
|
|
"""
|
|
Update configuration with new values
|
|
|
|
Args:
|
|
updates: Dictionary of updates (e.g., {"llm": {"api_key": "xxx"}})
|
|
"""
|
|
current = self.config.to_dict()
|
|
|
|
# Deep merge
|
|
def deep_merge(base: dict, updates: dict) -> dict:
|
|
for key, value in updates.items():
|
|
if key in base and isinstance(base[key], dict) and isinstance(value, dict):
|
|
deep_merge(base[key], value)
|
|
else:
|
|
base[key] = value
|
|
return base
|
|
|
|
merged = deep_merge(current, updates)
|
|
self.config = ReelForgeConfig(**merged)
|
|
|
|
def get(self, key: str, default: Any = None) -> Any:
|
|
"""Dict-like access (for backward compatibility)"""
|
|
return self.config.to_dict().get(key, default)
|
|
|
|
def validate(self) -> bool:
|
|
"""Validate configuration completeness"""
|
|
return self.config.validate_required()
|
|
|
|
def get_llm_config(self) -> dict:
|
|
"""Get LLM configuration as dict"""
|
|
return {
|
|
"api_key": self.config.llm.api_key,
|
|
"base_url": self.config.llm.base_url,
|
|
"model": self.config.llm.model,
|
|
}
|
|
|
|
def set_llm_config(self, api_key: str, base_url: str, model: str):
|
|
"""Set LLM configuration"""
|
|
self.update({
|
|
"llm": {
|
|
"api_key": api_key,
|
|
"base_url": base_url,
|
|
"model": model,
|
|
}
|
|
})
|
|
|
|
def get_image_config(self) -> dict:
|
|
"""Get image configuration as dict"""
|
|
return {
|
|
"default": self.config.image.default,
|
|
"comfyui_url": self.config.image.comfyui_url,
|
|
"runninghub_api_key": self.config.image.runninghub_api_key,
|
|
"prompt_prefix": self.config.image.prompt_prefix,
|
|
}
|
|
|
|
def set_image_config(
|
|
self,
|
|
comfyui_url: Optional[str] = None,
|
|
runninghub_api_key: Optional[str] = None
|
|
):
|
|
"""Set image configuration"""
|
|
updates = {}
|
|
if comfyui_url is not None:
|
|
updates["comfyui_url"] = comfyui_url
|
|
if runninghub_api_key is not None:
|
|
updates["runninghub_api_key"] = runninghub_api_key
|
|
|
|
if updates:
|
|
self.update({"image": updates})
|
|
|