lighting_analyzer.py

import numpy as np
import cv2
import logging
import traceback
from typing import Dict, Any, Optional

from configuration_manager import ConfigurationManager
from feature_extractor import FeatureExtractor
from indoor_outdoor_classifier import IndoorOutdoorClassifier
from lighting_condition_analyzer import LightingConditionAnalyzer


class LightingAnalyzer:
    """
    Comprehensive lighting analysis system facade that coordinates feature extraction,
    indoor/outdoor classification, and lighting condition determination.
    此class是一個總窗口，主要匯總各式光線分析相關的class

    This facade class maintains the original interface while internally delegating
    work to specialized components for improved maintainability and modularity.
    """

    def __init__(self, config: Optional[Dict[str, Any]] = None):
        """
        Initialize the lighting analyzer with configuration.

        Args:
            config: Optional configuration dictionary. If None, uses default configuration.
        """
        self.logger = self._setup_logger()

        try:
            # Initialize configuration manager
            self.config_manager = ConfigurationManager()

            # Override default configuration if provided
            if config is not None:
                self._update_configuration(config)

            # Initialize specialized components
            self.feature_extractor = FeatureExtractor(self.config_manager)
            self.indoor_outdoor_classifier = IndoorOutdoorClassifier(self.config_manager)
            self.lighting_condition_analyzer = LightingConditionAnalyzer(self.config_manager)

            # Legacy configuration access for backward compatibility
            self.config = self.config_manager.get_legacy_config_dict()

            self.logger.info("LightingAnalyzer initialized successfully")

        except Exception as e:
            self.logger.error(f"Error initializing LightingAnalyzer: {str(e)}")
            self.logger.error(f"Traceback: {traceback.format_exc()}")
            raise

    def _setup_logger(self) -> logging.Logger:
        """Set up logger for lighting analysis operations."""
        logger = logging.getLogger(f"{__name__}.LightingAnalyzer")
        if not logger.handlers:
            handler = logging.StreamHandler()
            formatter = logging.Formatter(
                '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
            )
            handler.setFormatter(formatter)
            logger.addHandler(handler)
            logger.setLevel(logging.INFO)
        return logger

    def _update_configuration(self, config: Dict[str, Any]) -> None:
        """
        Update configuration manager with provided configuration dictionary.

        Args:
            config: Configuration dictionary to update existing configuration.
        """
        try:
            # Update configuration through the manager's internal method
            self.config_manager._update_from_dict(config)
            self.logger.debug("Configuration updated successfully")

        except Exception as e:
            self.logger.warning(f"Error updating configuration: {str(e)}")
            self.logger.warning("Continuing with default configuration")

    def analyze(self, image, places365_info: Optional[Dict] = None) -> Dict[str, Any]:
        """
        Analyze lighting conditions of an image.

        This is the main entry point that maintains compatibility with the original interface
        while leveraging the new modular architecture internally.

        Args:
            image: Input image (numpy array or PIL Image).
            places365_info: Optional Places365 classification information containing
                          scene type, confidence, attributes, and indoor/outdoor classification.

        Returns:
            Dictionary containing comprehensive lighting analysis results including:
            - time_of_day: Specific lighting condition classification
            - confidence: Confidence score for the classification
            - is_indoor: Boolean indicating indoor/outdoor classification
            - indoor_probability: Probability score for indoor classification
            - brightness: Brightness analysis metrics
            - color_info: Color characteristic analysis
            - texture_info: Texture and gradient analysis
            - structure_info: Structural feature analysis
            - diagnostics: Detailed diagnostic information (if enabled)
        """
        try:
            self.logger.debug("Starting comprehensive lighting analysis")

            # Step 1: Validate and preprocess input image
            processed_image = self._preprocess_image(image)
            if processed_image is None:
                return self._get_error_result("Invalid image input")

            # Step 2: Extract comprehensive features
            self.logger.debug("Extracting image features")
            features = self.feature_extractor.extract_features(processed_image)

            if not features or "avg_brightness" not in features:
                return self._get_error_result("Feature extraction failed")

            # Step 3: Classify indoor/outdoor with Places365 integration
            self.logger.debug("Performing indoor/outdoor classification")
            indoor_outdoor_result = self.indoor_outdoor_classifier.classify(
                features, places365_info
            )

            is_indoor = indoor_outdoor_result["is_indoor"]
            indoor_probability = indoor_outdoor_result["indoor_probability"]

            # Step 4: Determine specific lighting conditions
            self.logger.debug(f"Analyzing lighting conditions for {'indoor' if is_indoor else 'outdoor'} scene")
            lighting_result = self.lighting_condition_analyzer.analyze_lighting_conditions(
                features, is_indoor, places365_info
            )

            # Step 5: Consolidate comprehensive results
            result = self._consolidate_analysis_results(
                lighting_result, indoor_outdoor_result, features
            )

            self.logger.info(f"Analysis complete: {result['time_of_day']} "
                           f"({'indoor' if result['is_indoor'] else 'outdoor'}) "
                           f"confidence: {result['confidence']:.3f}")

            return result

        except Exception as e:
            self.logger.error(f"Error in lighting analysis: {str(e)}")
            self.logger.error(f"Traceback: {traceback.format_exc()}")
            return self._get_error_result(str(e))

    def _preprocess_image(self, image) -> Optional[np.ndarray]:
        """
        Preprocess input image to ensure consistent format for analysis.

        Args:
            image: Input image in various possible formats.

        Returns:
            Preprocessed image as RGB numpy array, or None if preprocessing failed.
        """
        try:
            # Convert to numpy array if needed
            if not isinstance(image, np.ndarray):
                image_np = np.array(image)
            else:
                image_np = image.copy()

            # Validate basic image properties
            if len(image_np.shape) < 2:
                self.logger.error("Image must have at least 2 dimensions")
                return None

            height, width = image_np.shape[:2]
            if height == 0 or width == 0:
                self.logger.error(f"Invalid image dimensions: {height}x{width}")
                return None

            # Handle different color formats and convert to RGB
            if len(image_np.shape) == 2:
                # 灰階 to RGB
                image_rgb = cv2.cvtColor(image_np, cv2.COLOR_GRAY2RGB)
            elif image_np.shape[2] == 3:
                # Handle BGR vs RGB
                if not isinstance(image, np.ndarray):
                    # PIL images are typically RGB
                    image_rgb = image_np
                else:
                    # OpenCV arrays are typically BGR, convert to RGB
                    image_rgb = cv2.cvtColor(image_np, cv2.COLOR_BGR2RGB)
            elif image_np.shape[2] == 4:
                # RGBA to RGB
                if not isinstance(image, np.ndarray):
                    # PIL RGBA to RGB
                    image_rgb = cv2.cvtColor(image_np, cv2.COLOR_RGBA2RGB)
                else:
                    # OpenCV BGRA to RGB
                    image_rgb = cv2.cvtColor(image_np, cv2.COLOR_BGRA2RGB)
            else:
                self.logger.error(f"Unsupported image format with shape: {image_np.shape}")
                return None

            # Ensure uint8 data type
            if image_rgb.dtype != np.uint8:
                if image_rgb.dtype in [np.float32, np.float64]:
                    # Assume normalized float values
                    if image_rgb.max() <= 1.0:
                        image_rgb = (image_rgb * 255).astype(np.uint8)
                    else:
                        image_rgb = image_rgb.astype(np.uint8)
                else:
                    image_rgb = image_rgb.astype(np.uint8)

            self.logger.debug(f"Preprocessed image: {image_rgb.shape}, dtype: {image_rgb.dtype}")
            return image_rgb

        except Exception as e:
            self.logger.error(f"Error preprocessing image: {str(e)}")
            return None

    def _consolidate_analysis_results(self, lighting_result: Dict[str, Any],
                                     indoor_outdoor_result: Dict[str, Any],
                                     features: Dict[str, Any]) -> Dict[str, Any]:
        """
        Consolidate results from all analysis components into final output format.

        Args:
            lighting_result: Results from lighting condition analysis.
            indoor_outdoor_result: Results from indoor/outdoor classification.
            features: Extracted image features.

        Returns:
            Consolidated analysis results in the expected output format.
        """
        # Extract core results
        time_of_day = lighting_result["time_of_day"]
        confidence = lighting_result["confidence"]
        is_indoor = indoor_outdoor_result["is_indoor"]
        indoor_probability = indoor_outdoor_result["indoor_probability"]

        # Organize brightness information
        brightness_info = {
            "average": float(features.get("avg_brightness", 0.0)),
            "std_dev": float(features.get("brightness_std", 0.0)),
            "dark_ratio": float(features.get("dark_pixel_ratio", 0.0)),
            "bright_ratio": float(features.get("bright_pixel_ratio", 0.0))
        }

        # Organize color information
        color_info = {
            "blue_ratio": float(features.get("blue_ratio", 0.0)),
            "sky_like_blue_ratio": float(features.get("sky_like_blue_ratio", 0.0)),
            "yellow_orange_ratio": float(features.get("yellow_orange_ratio", 0.0)),
            "gray_ratio": float(features.get("gray_ratio", 0.0)),
            "avg_saturation": float(features.get("avg_saturation", 0.0)),
            "sky_region_brightness_ratio": float(features.get("sky_region_brightness_ratio", 1.0)),
            "sky_region_saturation": float(features.get("sky_region_saturation", 0.0)),
            "sky_region_blue_dominance": float(features.get("sky_region_blue_dominance", 0.0)),
            "color_atmosphere": features.get("color_atmosphere", "neutral"),
            "warm_ratio": float(features.get("warm_ratio", 0.0)),
            "cool_ratio": float(features.get("cool_ratio", 0.0))
        }

        # Organize texture information
        texture_info = {
            "gradient_ratio_vertical_horizontal": float(features.get("gradient_ratio_vertical_horizontal", 0.0)),
            "top_region_texture_complexity": float(features.get("top_region_texture_complexity", 0.0)),
            "shadow_clarity_score": float(features.get("shadow_clarity_score", 0.5))
        }

        # Organize structure information
        structure_info = {
            "ceiling_likelihood": float(features.get("ceiling_likelihood", 0.0)),
            "boundary_clarity": float(features.get("boundary_clarity", 0.0)),
            "openness_top_edge": float(features.get("openness_top_edge", 0.5))
        }

        # Compile final result
        result = {
            "time_of_day": time_of_day,
            "confidence": float(confidence),
            "is_indoor": is_indoor,
            "indoor_probability": float(indoor_probability),
            "brightness": brightness_info,
            "color_info": color_info,
            "texture_info": texture_info,
            "structure_info": structure_info
        }

        # Add diagnostic information if enabled
        if self.config_manager.algorithm_parameters.include_diagnostics:
            diagnostics = {}

            # Combine diagnostics from all components
            if "diagnostics" in lighting_result:
                diagnostics["lighting_diagnostics"] = lighting_result["diagnostics"]

            if "diagnostics" in indoor_outdoor_result:
                diagnostics["indoor_outdoor_diagnostics"] = indoor_outdoor_result["diagnostics"]

            if "feature_contributions" in indoor_outdoor_result:
                diagnostics["feature_contributions"] = indoor_outdoor_result["feature_contributions"]

            result["diagnostics"] = diagnostics

        return result

    def _get_error_result(self, error_message: str) -> Dict[str, Any]:
        """
        Generate standardized error result format.

        Args:
            error_message: Description of the error that occurred.

        Returns:
            Dictionary containing error result with safe default values.
        """
        return {
            "time_of_day": "unknown",
            "confidence": 0.0,
            "is_indoor": False,
            "indoor_probability": 0.5,
            "brightness": {
                "average": 100.0,
                "std_dev": 50.0,
                "dark_ratio": 0.0,
                "bright_ratio": 0.0
            },
            "color_info": {
                "blue_ratio": 0.0,
                "sky_like_blue_ratio": 0.0,
                "yellow_orange_ratio": 0.0,
                "gray_ratio": 0.0,
                "avg_saturation": 100.0,
                "sky_region_brightness_ratio": 1.0,
                "sky_region_saturation": 0.0,
                "sky_region_blue_dominance": 0.0,
                "color_atmosphere": "neutral",
                "warm_ratio": 0.0,
                "cool_ratio": 0.0
            },
            "texture_info": {
                "gradient_ratio_vertical_horizontal": 1.0,
                "top_region_texture_complexity": 0.5,
                "shadow_clarity_score": 0.5
            },
            "structure_info": {
                "ceiling_likelihood": 0.0,
                "boundary_clarity": 0.0,
                "openness_top_edge": 0.5
            },
            "error": error_message
        }

    def get_configuration(self) -> Dict[str, Any]:
        """
        Get current configuration as dictionary for backward compatibility.

        Returns:
            Dictionary containing all current configuration parameters.
        """
        return self.config_manager.get_legacy_config_dict()

    def update_configuration(self, config_updates: Dict[str, Any]) -> None:
        """
        Update configuration parameters.

        Args:
            config_updates: Dictionary containing configuration parameters to update.
        """
        try:
            self.config_manager._update_from_dict(config_updates)
            # Update legacy config reference
            self.config = self.config_manager.get_legacy_config_dict()
            self.logger.info("Configuration updated successfully")

        except Exception as e:
            self.logger.error(f"Error updating configuration: {str(e)}")
            raise

    def validate_configuration(self) -> bool:
        """
        Validate current configuration for logical consistency.

        Returns:
            True if configuration is valid, False otherwise.
        """
        try:
            validation_errors = self.config_manager.validate_configuration()

            if validation_errors:
                self.logger.error("Configuration validation failed:")
                for error in validation_errors:
                    self.logger.error(f"  - {error}")
                return False

            self.logger.info("Configuration validation passed")
            return True

        except Exception as e:
            self.logger.error(f"Error during configuration validation: {str(e)}")
            return False

    def save_configuration(self, filepath: str) -> None:
        """
        Save current configuration to file.

        Args:
            filepath: Path where to save the configuration file.
        """
        try:
            self.config_manager.save_to_file(filepath)
            self.logger.info(f"Configuration saved to {filepath}")

        except Exception as e:
            self.logger.error(f"Error saving configuration: {str(e)}")
            raise

    def load_configuration(self, filepath: str) -> None:
        """
        Load configuration from file.

        Args:
            filepath: Path to the configuration file to load.
        """
        try:
            self.config_manager.load_from_file(filepath)
            # Update legacy config reference
            self.config = self.config_manager.get_legacy_config_dict()
            self.logger.info(f"Configuration loaded from {filepath}")

        except Exception as e:
            self.logger.error(f"Error loading configuration: {str(e)}")
            raise