import numpy as np import logging import traceback from typing import List, Dict, Tuple, Optional, Union, Any from PIL import Image class ImageAnalyzer: """ 專注於圖像分析和預處理，包括多尺度金字塔分析、視角分析、建築特徵識別和圖像增強等功能 """ def __init__(self): """ 初始化圖像分析器 """ self.logger = logging.getLogger(__name__) def get_image_hash(self, image: Union[Image.Image, np.ndarray]) -> int: """ 為圖像生成簡單的 hash 值用於快取 Args: image: PIL Image 或 numpy 數組 Returns: int: 圖像的 hash 值 """ try: if isinstance(image, np.ndarray): # 對於 numpy 數組，降採樣並計算簡單 hash small_img = image[::10, ::10] if image.ndim == 3 else image return hash(small_img.tobytes()) else: # 對於 PIL 圖像，調整大小後轉換為 bytes small_img = image.resize((32, 32)) return hash(small_img.tobytes()) except Exception as e: self.logger.error(f"Error generating image hash: {e}") self.logger.error(traceback.format_exc()) return 0 def enhance_features(self, image: Union[Image.Image, np.ndarray]) -> Image.Image: """ 增強圖像特徵以改善地標檢測 Args: image: 輸入圖像 Returns: PIL.Image: 增強後的圖像 """ try: # ensure PIL format if not isinstance(image, Image.Image): if isinstance(image, np.ndarray): image = Image.fromarray(image) else: raise ValueError("Unsupported image format. Expected PIL Image or numpy array.") # 轉換為numpy進行處理 img_array = np.array(image) # 跳過灰度圖像的處理 if len(img_array.shape) < 3: return image # 應用自適應對比度增強 try: from skimage import color, exposure # 轉換到LAB色彩空間 if img_array.shape[2] == 4: # 處理RGBA img_array = img_array[:,:,:3] lab = color.rgb2lab(img_array[:,:,:3] / 255.0) l_channel = lab[:,:,0] # 增強L通道的對比度 p2, p98 = np.percentile(l_channel, (2, 98)) l_channel_enhanced = exposure.rescale_intensity(l_channel, in_range=(p2, p98)) # 替換L通道並轉換回RGB lab[:,:,0] = l_channel_enhanced enhanced_img = color.lab2rgb(lab) * 255.0 enhanced_img = enhanced_img.astype(np.uint8) return Image.fromarray(enhanced_img) except ImportError: self.logger.warning("skimage not available for feature enhancement") return image except Exception as e: self.logger.error(f"Error in feature enhancement: {e}") self.logger.error(traceback.format_exc()) return image def analyze_viewpoint(self, image: Union[Image.Image, np.ndarray], clip_model_manager) -> Dict[str, Any]: """ 分析圖像視角以調整檢測參數 Args: image: 輸入圖像 clip_model_manager: CLIP模型管理器實例 Returns: Dict: 視角分析結果 """ try: viewpoint_prompts = { "aerial_view": "an aerial view from above looking down", "street_level": "a street level view looking up at a tall structure", "eye_level": "an eye-level horizontal view of a landmark", "distant": "a distant view of a landmark on the horizon", "close_up": "a close-up detailed view of architectural features", "interior": "an interior view inside a structure", "angled_view": "an angled view of a structure", "low_angle": "a low angle view looking up at a building" } # 計算相似度分數 viewpoint_scores = self.calculate_similarity_scores(image, viewpoint_prompts, clip_model_manager) # 找到主要視角 dominant_viewpoint = max(viewpoint_scores.items(), key=lambda x: x[1]) return { "viewpoint_scores": viewpoint_scores, "dominant_viewpoint": dominant_viewpoint[0], "confidence": dominant_viewpoint[1] } except Exception as e: self.logger.error(f"Error in viewpoint analysis: {e}") self.logger.error(traceback.format_exc()) return { "viewpoint_scores": {}, "dominant_viewpoint": "eye_level", "confidence": 0.0 } def calculate_similarity_scores(self, image: Union[Image.Image, np.ndarray], prompts: Dict[str, str], clip_model_manager) -> Dict[str, float]: """ 計算圖像與一組特定提示之間的相似度分數 Args: image: 輸入圖像 prompts: 提示詞字典 {名稱: 提示文本} clip_model_manager: CLIP模型管理器實例 Returns: Dict[str, float]: 每個提示的相似度分數 """ try: # ensure PIL format if not isinstance(image, Image.Image): if isinstance(image, np.ndarray): image = Image.fromarray(image) else: raise ValueError("Unsupported image format. Expected PIL Image or numpy array.") # preprocess image image_input = clip_model_manager.preprocess_image(image) # get image features image_features = clip_model_manager.encode_image(image_input) # 計算與每個提示的similarity scores = {} prompt_texts = list(prompts.values()) prompt_features = clip_model_manager.encode_single_text(prompt_texts) # 計算相似度 similarity = clip_model_manager.calculate_similarity(image_features, prompt_features) # result for i, (name, _) in enumerate(prompts.items()): scores[name] = float(similarity[0][i]) return scores except Exception as e: self.logger.error(f"Error calculating similarity scores: {e}") self.logger.error(traceback.format_exc()) return {} def analyze_architectural_features(self, image: Union[Image.Image, np.ndarray], clip_model_manager) -> Dict[str, Any]: """ 分析圖像中結構的建築特徵，不硬編碼特定地標 Args: image: 輸入圖像 clip_model_manager: CLIP模型管理器實例 Returns: Dict: 建築特徵分析結果 """ try: # 定義通用建築特徵提示，適用於所有類型的地標 architecture_prompts = { "tall_structure": "a tall vertical structure standing alone", "tiered_building": "a building with multiple stacked tiers or segments", "historical_structure": "a building with historical architectural elements", "modern_design": "a modern structure with contemporary architectural design", "segmented_exterior": "a structure with visible segmented or sectioned exterior", "viewing_platform": "a tall structure with observation area at the top", "time_display": "a structure with timepiece features", "glass_facade": "a building with prominent glass exterior surfaces", "memorial_structure": "a monument or memorial structure", "ancient_construction": "ancient constructed elements or archaeological features", "natural_landmark": "a natural geographic formation or landmark", "slanted_design": "a structure with non-vertical or leaning profile" } # 計算與通用建築模式的相似度分數 context_scores = self.calculate_similarity_scores(image, architecture_prompts, clip_model_manager) # 確定最相關的建築特徵 top_features = sorted(context_scores.items(), key=lambda x: x[1], reverse=True)[:3] # 計算特徵置信度 context_confidence = sum(score for _, score in top_features) / 3 # 根據頂級特徵確定主要建築類別 architectural_categories = { "tower": ["tall_structure", "viewing_platform", "time_display"], "skyscraper": ["tall_structure", "modern_design", "glass_facade"], "historical": ["historical_structure", "ancient_construction", "memorial_structure"], "natural": ["natural_landmark"], "distinctive": ["tiered_building", "segmented_exterior", "slanted_design"] } # 根據頂級特徵為每個類別評分 category_scores = {} for category, features in architectural_categories.items(): category_score = 0 for feature, score in context_scores.items(): if feature in features: category_score += score category_scores[category] = category_score primary_category = max(category_scores.items(), key=lambda x: x[1])[0] return { "architectural_features": top_features, "context_confidence": context_confidence, "primary_category": primary_category, "category_scores": category_scores } except Exception as e: self.logger.error(f"Error in architectural feature analysis: {e}") self.logger.error(traceback.format_exc()) return { "architectural_features": [], "context_confidence": 0.0, "primary_category": "building", "category_scores": {} } def perform_pyramid_analysis(self, image: Union[Image.Image, np.ndarray], clip_model_manager, landmark_data_manager, levels: int = 4, base_threshold: float = 0.25, aspect_ratios: List[float] = [1.0, 0.75, 1.5]) -> Dict[str, Any]: """ 對圖像執行多尺度金字塔分析以改善地標檢測 Args: image: 輸入圖像 clip_model_manager: CLIP模型管理器實例 landmark_data_manager: 地標數據管理器實例 levels: 金字塔層級數 base_threshold: 基礎置信度閾值 aspect_ratios: 不同縱橫比列表 Returns: Dict: 金字塔分析結果 """ try: # 確保圖像是PIL格式 if not isinstance(image, Image.Image): if isinstance(image, np.ndarray): image = Image.fromarray(image) else: raise ValueError("Unsupported image format. Expected PIL Image or numpy array.") width, height = image.size pyramid_results = [] # 獲取預計算的地標文本特徵 landmark_prompts = landmark_data_manager.get_landmark_prompts() if not landmark_prompts: return { "is_landmark": False, "results": [], "best_result": None } landmark_text_features = clip_model_manager.encode_text_batch(landmark_prompts) # 對每個縮放和縱橫比組合進行處理 for level in range(levels): # 計算縮放因子 scale_factor = 1.0 - (level * 0.2) for aspect_ratio in aspect_ratios: # 計算新尺寸，保持面積近似不變 if aspect_ratio != 1.0: # 保持面積近似不變的情況下調整縱橫比 new_width = int(width * scale_factor * (1/aspect_ratio)**0.5) new_height = int(height * scale_factor * aspect_ratio**0.5) else: new_width = int(width * scale_factor) new_height = int(height * scale_factor) # 調整圖像大小 scaled_image = image.resize((new_width, new_height), Image.LANCZOS) # 預處理圖像 image_input = clip_model_manager.preprocess_image(scaled_image) # 獲取圖像特徵 image_features = clip_model_manager.encode_image(image_input) # 計算相似度 similarity = clip_model_manager.calculate_similarity(image_features, landmark_text_features) # 找到最佳匹配 best_idx = similarity[0].argmax().item() best_score = similarity[0][best_idx] if best_score >= base_threshold: landmark_id, landmark_info = landmark_data_manager.get_landmark_by_index(best_idx) if landmark_id: pyramid_results.append({ "landmark_id": landmark_id, "landmark_name": landmark_info.get("name", "Unknown"), "confidence": float(best_score), "scale_factor": scale_factor, "aspect_ratio": aspect_ratio, "location": landmark_info.get("location", "Unknown Location") }) # 按置信度排序 pyramid_results.sort(key=lambda x: x["confidence"], reverse=True) return { "is_landmark": len(pyramid_results) > 0, "results": pyramid_results, "best_result": pyramid_results[0] if pyramid_results else None } except Exception as e: self.logger.error(f"Error in pyramid analysis: {e}") self.logger.error(traceback.format_exc()) return { "is_landmark": False, "results": [], "best_result": None }