"""Application configuration management.""" import os import logging from typing import Dict, List, Optional, Any from dataclasses import dataclass, field from pathlib import Path logger = logging.getLogger(__name__) @dataclass class TTSConfig: """Configuration for TTS providers.""" preferred_providers: List[str] = field(default_factory=lambda: ['chatterbox', 'dummy']) default_voice: str = 'default' default_speed: float = 1.0 default_language: str = 'en' enable_streaming: bool = True max_text_length: int = 10000 @dataclass class STTConfig: """Configuration for STT providers.""" preferred_providers: List[str] = field(default_factory=lambda: ['parakeet', 'whisper']) default_model: str = 'parakeet' chunk_length_s: int = 30 batch_size: int = 16 enable_vad: bool = True @dataclass class TranslationConfig: """Configuration for translation providers.""" default_provider: str = 'nllb' model_name: str = 'facebook/nllb-200-3.3B' max_chunk_length: int = 1000 batch_size: int = 8 cache_translations: bool = True @dataclass class ProcessingConfig: """Configuration for audio processing pipeline.""" temp_dir: str = '/tmp/audio_processing' cleanup_temp_files: bool = True max_file_size_mb: int = 100 supported_audio_formats: List[str] = field(default_factory=lambda: ['wav', 'mp3', 'flac', 'ogg']) processing_timeout_seconds: int = 300 @dataclass class LoggingConfig: """Configuration for logging.""" level: str = 'INFO' format: str = '%(asctime)s - %(name)s - %(levelname)s - %(message)s' enable_file_logging: bool = False log_file_path: str = 'app.log' max_log_file_size_mb: int = 10 backup_count: int = 5 class AppConfig: """Centralized application configuration management.""" def __init__(self, config_file: Optional[str] = None): """ Initialize application configuration. Args: config_file: Optional path to configuration file """ self.config_file = config_file self._config_data: Dict[str, Any] = {} # Initialize configuration sections self.tts = TTSConfig() self.stt = STTConfig() self.translation = TranslationConfig() self.processing = ProcessingConfig() self.logging = LoggingConfig() # Load configuration self._load_configuration() def _load_configuration(self) -> None: """Load configuration from environment variables and config file.""" try: # Load from environment variables first self._load_from_environment() # Load from config file if provided if self.config_file and os.path.exists(self.config_file): self._load_from_file() # Validate configuration self._validate_configuration() logger.info("Configuration loaded successfully") except Exception as e: logger.error(f"Failed to load configuration: {e}") # Use default configuration logger.info("Using default configuration") def _load_from_environment(self) -> None: """Load configuration from environment variables.""" # TTS Configuration if os.getenv('TTS_PREFERRED_PROVIDERS'): self.tts.preferred_providers = os.getenv('TTS_PREFERRED_PROVIDERS').split(',') if os.getenv('TTS_DEFAULT_VOICE'): self.tts.default_voice = os.getenv('TTS_DEFAULT_VOICE') if os.getenv('TTS_DEFAULT_SPEED'): self.tts.default_speed = float(os.getenv('TTS_DEFAULT_SPEED')) if os.getenv('TTS_DEFAULT_LANGUAGE'): self.tts.default_language = os.getenv('TTS_DEFAULT_LANGUAGE') if os.getenv('TTS_ENABLE_STREAMING'): self.tts.enable_streaming = os.getenv('TTS_ENABLE_STREAMING').lower() == 'true' # STT Configuration if os.getenv('STT_PREFERRED_PROVIDERS'): self.stt.preferred_providers = os.getenv('STT_PREFERRED_PROVIDERS').split(',') if os.getenv('STT_DEFAULT_MODEL'): self.stt.default_model = os.getenv('STT_DEFAULT_MODEL') if os.getenv('STT_CHUNK_LENGTH'): self.stt.chunk_length_s = int(os.getenv('STT_CHUNK_LENGTH')) if os.getenv('STT_BATCH_SIZE'): self.stt.batch_size = int(os.getenv('STT_BATCH_SIZE')) # Translation Configuration if os.getenv('TRANSLATION_DEFAULT_PROVIDER'): self.translation.default_provider = os.getenv('TRANSLATION_DEFAULT_PROVIDER') if os.getenv('TRANSLATION_MODEL_NAME'): self.translation.model_name = os.getenv('TRANSLATION_MODEL_NAME') if os.getenv('TRANSLATION_MAX_CHUNK_LENGTH'): self.translation.max_chunk_length = int(os.getenv('TRANSLATION_MAX_CHUNK_LENGTH')) # Processing Configuration if os.getenv('PROCESSING_TEMP_DIR'): self.processing.temp_dir = os.getenv('PROCESSING_TEMP_DIR') if os.getenv('PROCESSING_CLEANUP_TEMP_FILES'): self.processing.cleanup_temp_files = os.getenv('PROCESSING_CLEANUP_TEMP_FILES').lower() == 'true' if os.getenv('PROCESSING_MAX_FILE_SIZE_MB'): self.processing.max_file_size_mb = int(os.getenv('PROCESSING_MAX_FILE_SIZE_MB')) if os.getenv('PROCESSING_TIMEOUT_SECONDS'): self.processing.processing_timeout_seconds = int(os.getenv('PROCESSING_TIMEOUT_SECONDS')) # Logging Configuration if os.getenv('LOG_LEVEL'): self.logging.level = os.getenv('LOG_LEVEL') if os.getenv('LOG_FORMAT'): self.logging.format = os.getenv('LOG_FORMAT') if os.getenv('LOG_FILE_PATH'): self.logging.log_file_path = os.getenv('LOG_FILE_PATH') def _load_from_file(self) -> None: """Load configuration from file (JSON or YAML).""" try: import json with open(self.config_file, 'r') as f: if self.config_file.endswith('.json'): self._config_data = json.load(f) elif self.config_file.endswith(('.yml', '.yaml')): try: import yaml self._config_data = yaml.safe_load(f) except ImportError: logger.warning("PyYAML not installed, cannot load YAML config file") return else: logger.warning(f"Unsupported config file format: {self.config_file}") return # Apply configuration from file self._apply_config_data() except Exception as e: logger.error(f"Failed to load config file {self.config_file}: {e}") def _apply_config_data(self) -> None: """Apply configuration data from loaded file.""" if not self._config_data: return # Apply TTS configuration tts_config = self._config_data.get('tts', {}) if 'preferred_providers' in tts_config: self.tts.preferred_providers = tts_config['preferred_providers'] if 'default_voice' in tts_config: self.tts.default_voice = tts_config['default_voice'] if 'default_speed' in tts_config: self.tts.default_speed = tts_config['default_speed'] if 'default_language' in tts_config: self.tts.default_language = tts_config['default_language'] if 'enable_streaming' in tts_config: self.tts.enable_streaming = tts_config['enable_streaming'] # Apply STT configuration stt_config = self._config_data.get('stt', {}) if 'preferred_providers' in stt_config: self.stt.preferred_providers = stt_config['preferred_providers'] if 'default_model' in stt_config: self.stt.default_model = stt_config['default_model'] if 'chunk_length_s' in stt_config: self.stt.chunk_length_s = stt_config['chunk_length_s'] if 'batch_size' in stt_config: self.stt.batch_size = stt_config['batch_size'] # Apply Translation configuration translation_config = self._config_data.get('translation', {}) if 'default_provider' in translation_config: self.translation.default_provider = translation_config['default_provider'] if 'model_name' in translation_config: self.translation.model_name = translation_config['model_name'] if 'max_chunk_length' in translation_config: self.translation.max_chunk_length = translation_config['max_chunk_length'] # Apply Processing configuration processing_config = self._config_data.get('processing', {}) if 'temp_dir' in processing_config: self.processing.temp_dir = processing_config['temp_dir'] if 'cleanup_temp_files' in processing_config: self.processing.cleanup_temp_files = processing_config['cleanup_temp_files'] if 'max_file_size_mb' in processing_config: self.processing.max_file_size_mb = processing_config['max_file_size_mb'] if 'processing_timeout_seconds' in processing_config: self.processing.processing_timeout_seconds = processing_config['processing_timeout_seconds'] # Apply Logging configuration logging_config = self._config_data.get('logging', {}) if 'level' in logging_config: self.logging.level = logging_config['level'] if 'format' in logging_config: self.logging.format = logging_config['format'] if 'log_file_path' in logging_config: self.logging.log_file_path = logging_config['log_file_path'] def _validate_configuration(self) -> None: """Validate configuration values.""" # Validate TTS configuration if not (0.1 <= self.tts.default_speed <= 3.0): logger.warning(f"Invalid TTS speed {self.tts.default_speed}, using default 1.0") self.tts.default_speed = 1.0 if self.tts.max_text_length <= 0: logger.warning(f"Invalid max text length {self.tts.max_text_length}, using default 10000") self.tts.max_text_length = 10000 # Validate STT configuration if self.stt.chunk_length_s <= 0: logger.warning(f"Invalid chunk length {self.stt.chunk_length_s}, using default 30") self.stt.chunk_length_s = 30 if self.stt.batch_size <= 0: logger.warning(f"Invalid batch size {self.stt.batch_size}, using default 16") self.stt.batch_size = 16 # Validate Translation configuration if self.translation.max_chunk_length <= 0: logger.warning(f"Invalid max chunk length {self.translation.max_chunk_length}, using default 1000") self.translation.max_chunk_length = 1000 # Validate Processing configuration if self.processing.max_file_size_mb <= 0: logger.warning(f"Invalid max file size {self.processing.max_file_size_mb}, using default 100") self.processing.max_file_size_mb = 100 if self.processing.processing_timeout_seconds <= 0: logger.warning(f"Invalid timeout {self.processing.processing_timeout_seconds}, using default 300") self.processing.processing_timeout_seconds = 300 # Ensure temp directory exists try: Path(self.processing.temp_dir).mkdir(parents=True, exist_ok=True) except Exception as e: logger.warning(f"Failed to create temp directory {self.processing.temp_dir}: {e}") self.processing.temp_dir = '/tmp/audio_processing' Path(self.processing.temp_dir).mkdir(parents=True, exist_ok=True) def get_tts_config(self) -> Dict[str, Any]: """Get TTS configuration as dictionary.""" return { 'preferred_providers': self.tts.preferred_providers, 'default_voice': self.tts.default_voice, 'default_speed': self.tts.default_speed, 'default_language': self.tts.default_language, 'enable_streaming': self.tts.enable_streaming, 'max_text_length': self.tts.max_text_length } def get_stt_config(self) -> Dict[str, Any]: """Get STT configuration as dictionary.""" return { 'preferred_providers': self.stt.preferred_providers, 'default_model': self.stt.default_model, 'chunk_length_s': self.stt.chunk_length_s, 'batch_size': self.stt.batch_size, 'enable_vad': self.stt.enable_vad } def get_translation_config(self) -> Dict[str, Any]: """Get translation configuration as dictionary.""" return { 'default_provider': self.translation.default_provider, 'model_name': self.translation.model_name, 'max_chunk_length': self.translation.max_chunk_length, 'batch_size': self.translation.batch_size, 'cache_translations': self.translation.cache_translations } def get_processing_config(self) -> Dict[str, Any]: """Get processing configuration as dictionary.""" return { 'temp_dir': self.processing.temp_dir, 'cleanup_temp_files': self.processing.cleanup_temp_files, 'max_file_size_mb': self.processing.max_file_size_mb, 'supported_audio_formats': self.processing.supported_audio_formats, 'processing_timeout_seconds': self.processing.processing_timeout_seconds } def get_logging_config(self) -> Dict[str, Any]: """Get logging configuration as dictionary.""" return { 'level': self.logging.level, 'format': self.logging.format, 'enable_file_logging': self.logging.enable_file_logging, 'log_file_path': self.logging.log_file_path, 'max_log_file_size_mb': self.logging.max_log_file_size_mb, 'backup_count': self.logging.backup_count } def reload_configuration(self) -> None: """Reload configuration from sources.""" logger.info("Reloading configuration") self._load_configuration() def save_configuration(self, output_file: str) -> None: """ Save current configuration to file. Args: output_file: Path to output configuration file """ try: config_dict = { 'tts': self.get_tts_config(), 'stt': self.get_stt_config(), 'translation': self.get_translation_config(), 'processing': self.get_processing_config(), 'logging': self.get_logging_config() } import json with open(output_file, 'w') as f: json.dump(config_dict, f, indent=2) logger.info(f"Configuration saved to {output_file}") except Exception as e: logger.error(f"Failed to save configuration to {output_file}: {e}") raise def __str__(self) -> str: """String representation of configuration.""" return ( f"AppConfig(\n" f" TTS: {self.tts}\n" f" STT: {self.stt}\n" f" Translation: {self.translation}\n" f" Processing: {self.processing}\n" f" Logging: {self.logging}\n" f")" )