""" Application Configuration Module Centralizes all application configuration and environment variable handling Note: Named 'app_config.py' to avoid conflicts with existing 'Configs/' folder """ import os from dataclasses import dataclass, asdict from typing import Dict, Any, Optional import logging logger = logging.getLogger(__name__) @dataclass class ProcessingConfig: """ Main processing configuration with environment variable defaults """ # Frame processing settings keyframe_interval: int = int(os.getenv('KEYFRAME_INTERVAL', '5')) frame_skip: int = int(os.getenv('FRAME_SKIP', '1')) # Memory management memory_cleanup_interval: int = int(os.getenv('MEMORY_CLEANUP_INTERVAL', '30')) # Video constraints max_video_length: int = int(os.getenv('MAX_VIDEO_LENGTH', '300')) # seconds max_video_resolution: str = os.getenv('MAX_VIDEO_RESOLUTION', '1920x1080') # Quality settings quality_preset: str = os.getenv('QUALITY_PRESET', 'balanced') # Model settings sam2_model_size: str = os.getenv('SAM2_MODEL_SIZE', 'large') # tiny, small, base, large matanyone_precision: str = os.getenv('MATANYONE_PRECISION', 'fp32') # fp16, fp32 # Processing settings temporal_consistency: bool = os.getenv('TEMPORAL_CONSISTENCY', 'true').lower() == 'true' edge_refinement: bool = os.getenv('EDGE_REFINEMENT', 'true').lower() == 'true' # Output settings output_format: str = os.getenv('OUTPUT_FORMAT', 'mp4') output_quality: str = os.getenv('OUTPUT_QUALITY', 'high') # low, medium, high preserve_audio: bool = os.getenv('PRESERVE_AUDIO', 'true').lower() == 'true' # Cache settings model_cache_dir: str = os.getenv('MODEL_CACHE_DIR', '/tmp/model_cache') temp_dir: str = os.getenv('TEMP_DIR', '/tmp') cleanup_temp_files: bool = os.getenv('CLEANUP_TEMP_FILES', 'true').lower() == 'true' # Performance settings max_concurrent_processes: int = int(os.getenv('MAX_CONCURRENT_PROCESSES', '1')) gpu_memory_fraction: float = float(os.getenv('GPU_MEMORY_FRACTION', '0.8')) # Debug settings debug_mode: bool = os.getenv('DEBUG_MODE', 'false').lower() == 'true' save_intermediate_results: bool = os.getenv('SAVE_INTERMEDIATE_RESULTS', 'false').lower() == 'true' def __post_init__(self): """Validate configuration after initialization""" self._validate_config() self._create_directories() if self.debug_mode: self._log_config() def _validate_config(self): """Validate configuration values""" # Validate frame settings if self.keyframe_interval < 1: logger.warning(f"keyframe_interval must be >= 1, got {self.keyframe_interval}. Setting to 1.") self.keyframe_interval = 1 if self.frame_skip < 1: logger.warning(f"frame_skip must be >= 1, got {self.frame_skip}. Setting to 1.") self.frame_skip = 1 # Validate memory settings if self.memory_cleanup_interval < 1: logger.warning(f"memory_cleanup_interval must be >= 1, got {self.memory_cleanup_interval}. Setting to 10.") self.memory_cleanup_interval = 10 # Validate video constraints if self.max_video_length < 1: logger.warning(f"max_video_length must be >= 1, got {self.max_video_length}. Setting to 60.") self.max_video_length = 60 # Validate resolution format if 'x' not in self.max_video_resolution: logger.warning(f"Invalid resolution format: {self.max_video_resolution}. Setting to 1920x1080.") self.max_video_resolution = '1920x1080' # Validate quality preset valid_presets = ['fast', 'balanced', 'high', 'ultra'] if self.quality_preset not in valid_presets: logger.warning(f"Invalid quality preset: {self.quality_preset}. Setting to 'balanced'.") self.quality_preset = 'balanced' # Validate model settings valid_sam2_sizes = ['tiny', 'small', 'base', 'large'] if self.sam2_model_size not in valid_sam2_sizes: logger.warning(f"Invalid SAM2 model size: {self.sam2_model_size}. Setting to 'large'.") self.sam2_model_size = 'large' valid_precisions = ['fp16', 'fp32'] if self.matanyone_precision not in valid_precisions: logger.warning(f"Invalid precision: {self.matanyone_precision}. Setting to 'fp32'.") self.matanyone_precision = 'fp32' # Validate output settings valid_formats = ['mp4', 'avi', 'mov', 'webm'] if self.output_format not in valid_formats: logger.warning(f"Invalid output format: {self.output_format}. Setting to 'mp4'.") self.output_format = 'mp4' valid_qualities = ['low', 'medium', 'high'] if self.output_quality not in valid_qualities: logger.warning(f"Invalid output quality: {self.output_quality}. Setting to 'high'.") self.output_quality = 'high' # Validate performance settings if self.max_concurrent_processes < 1: logger.warning(f"max_concurrent_processes must be >= 1, got {self.max_concurrent_processes}. Setting to 1.") self.max_concurrent_processes = 1 if not (0.1 <= self.gpu_memory_fraction <= 1.0): logger.warning(f"gpu_memory_fraction must be between 0.1 and 1.0, got {self.gpu_memory_fraction}. Setting to 0.8.") self.gpu_memory_fraction = 0.8 def _create_directories(self): """Create necessary directories if they don't exist""" import os directories = [self.model_cache_dir, self.temp_dir] for directory in directories: try: os.makedirs(directory, exist_ok=True) logger.debug(f"Ensured directory exists: {directory}") except Exception as e: logger.error(f"Failed to create directory {directory}: {e}") def _log_config(self): """Log current configuration in debug mode""" logger.info("=== Processing Configuration ===") for key, value in asdict(self).items(): logger.info(f"{key}: {value}") logger.info("===============================") def to_dict(self) -> Dict[str, Any]: """Convert configuration to dictionary""" return asdict(self) def get_quality_settings(self) -> Dict[str, Any]: """Get quality-specific settings based on preset""" quality_maps = { 'fast': { 'keyframe_interval': max(self.keyframe_interval, 10), 'frame_skip': max(self.frame_skip, 2), 'edge_refinement': False, 'temporal_consistency': False, 'model_precision': 'fp16' }, 'balanced': { 'keyframe_interval': self.keyframe_interval, 'frame_skip': self.frame_skip, 'edge_refinement': True, 'temporal_consistency': True, 'model_precision': 'fp32' }, 'high': { 'keyframe_interval': max(self.keyframe_interval // 2, 1), 'frame_skip': 1, 'edge_refinement': True, 'temporal_consistency': True, 'model_precision': 'fp32' }, 'ultra': { 'keyframe_interval': 1, 'frame_skip': 1, 'edge_refinement': True, 'temporal_consistency': True, 'model_precision': 'fp32' } } return quality_maps.get(self.quality_preset, quality_maps['balanced']) def get_resolution_limits(self) -> tuple[int, int]: """Get max width and height from resolution setting""" try: width, height = map(int, self.max_video_resolution.split('x')) return width, height except ValueError: logger.error(f"Invalid resolution format: {self.max_video_resolution}") return 1920, 1080 def is_high_performance_mode(self) -> bool: """Check if configuration is set for high performance""" return ( self.quality_preset in ['high', 'ultra'] and self.edge_refinement and self.temporal_consistency and self.keyframe_interval <= 3 ) def get_memory_limits(self) -> Dict[str, float]: """Get memory-related limits""" return { 'gpu_memory_fraction': self.gpu_memory_fraction, 'cleanup_interval': self.memory_cleanup_interval, 'max_concurrent': self.max_concurrent_processes } # Singleton instance for application-wide use _config_instance: Optional[ProcessingConfig] = None def get_config() -> ProcessingConfig: """Get global configuration instance""" global _config_instance if _config_instance is None: _config_instance = ProcessingConfig() return _config_instance def reload_config() -> ProcessingConfig: """Reload configuration from environment variables""" global _config_instance _config_instance = ProcessingConfig() logger.info("Configuration reloaded from environment variables") return _config_instance def update_config(**kwargs) -> ProcessingConfig: """Update configuration with new values""" global _config_instance if _config_instance is None: _config_instance = ProcessingConfig() for key, value in kwargs.items(): if hasattr(_config_instance, key): setattr(_config_instance, key, value) logger.debug(f"Updated config: {key} = {value}") else: logger.warning(f"Unknown configuration key: {key}") # Re-validate after updates _config_instance._validate_config() return _config_instance