Create app_config.py

app_config.py

@@ -0,0 +1,246 @@
+"""
+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